Revision 4a105ce2 snf-pithos-backend/pithos/backends/lib/sqlalchemy/public.py
| b/snf-pithos-backend/pithos/backends/lib/sqlalchemy/public.py | ||
|---|---|---|
| 37 | 37 |
from sqlalchemy.schema import Index |
| 38 | 38 |
from sqlalchemy.exc import NoSuchTableError |
| 39 | 39 |
|
| 40 |
from pithos.backends.random_word import get_word |
|
| 40 |
from pithos.backends.random_word import get_random_word |
|
| 41 |
|
|
| 42 |
import logging |
|
| 43 |
|
|
| 44 |
logger = logging.getLogger(__name__) |
|
| 41 | 45 |
|
| 42 | 46 |
def create_tables(engine): |
| 43 | 47 |
metadata = MetaData() |
| ... | ... | |
| 68 | 72 |
tables = create_tables(self.engine) |
| 69 | 73 |
map(lambda t: self.__setattr__(t.name, t), tables) |
| 70 | 74 |
|
| 71 |
def get_unique_url(self, serial, public_url_min_length, public_url_alphabet):
|
|
| 72 |
l = public_url_min_length
|
|
| 75 |
def get_unique_url(self, public_security, public_url_alphabet):
|
|
| 76 |
l = public_security
|
|
| 73 | 77 |
while 1: |
| 74 |
candidate = get_word(serial, length=l, alphabet=public_url_alphabet)
|
|
| 78 |
candidate = get_random_word(length=l, alphabet=public_url_alphabet)
|
|
| 75 | 79 |
if self.public_path(candidate) is None: |
| 76 | 80 |
return candidate |
| 77 | 81 |
l +=1 |
| 78 | 82 |
|
| 79 |
def public_set(self, path, public_url_min_length, public_url_alphabet):
|
|
| 83 |
def public_set(self, path, public_security, public_url_alphabet):
|
|
| 80 | 84 |
s = select([self.public.c.public_id]) |
| 81 | 85 |
s = s.where(self.public.c.path == path) |
| 82 | 86 |
r = self.conn.execute(s) |
| ... | ... | |
| 84 | 88 |
r.close() |
| 85 | 89 |
|
| 86 | 90 |
if not row: |
| 91 |
url = self.get_unique_url( |
|
| 92 |
public_security, public_url_alphabet |
|
| 93 |
) |
|
| 87 | 94 |
s = self.public.insert() |
| 88 |
s = s.values(path=path, active=True) |
|
| 95 |
s = s.values(path=path, active=True, url=url)
|
|
| 89 | 96 |
r = self.conn.execute(s) |
| 90 |
serial = r.inserted_primary_key[0] |
|
| 91 | 97 |
r.close() |
| 92 |
|
|
| 93 |
url = self.get_unique_url( |
|
| 94 |
serial, public_url_min_length, public_url_alphabet |
|
| 95 |
) |
|
| 96 |
s = self.public.update().where(self.public.c.public_id == serial) |
|
| 97 |
s = s.values(url=url) |
|
| 98 |
self.conn.execute(s).close() |
|
| 98 |
logger.info('Public url: %s set for path: %s' % (url, path))
|
|
| 99 | 99 |
|
| 100 | 100 |
def public_unset(self, path): |
| 101 | 101 |
s = self.public.delete() |
| 102 | 102 |
s = s.where(self.public.c.path == path) |
| 103 | 103 |
self.conn.execute(s).close() |
| 104 |
logger.info('Public url unset for path: %s' % (path))
|
|
| 104 | 105 |
|
| 105 | 106 |
def public_unset_bulk(self, paths): |
| 106 | 107 |
if not paths: |
Also available in: Unified diff