Changeset View
Changeset View
Standalone View
Standalone View
swh/web/common/service.py
Show First 20 Lines • Show All 82 Lines • ▼ Show 20 Lines | def lookup_hash(q): | ||||
Args: query string of the form <hash_algo:hash> | Args: query string of the form <hash_algo:hash> | ||||
Returns: Dict with key found containing the hash info if the | Returns: Dict with key found containing the hash info if the | ||||
hash is present, None if not. | hash is present, None if not. | ||||
""" | """ | ||||
algo, hash = query.parse_hash(q) | algo, hash = query.parse_hash(q) | ||||
found = storage.content_find({algo: hash}) | found = _first_element(storage.content_find({algo: hash})) | ||||
return {'found': converters.from_content(found), | return {'found': converters.from_content(found), | ||||
'algo': algo} | 'algo': algo} | ||||
def search_hash(q): | def search_hash(q): | ||||
"""Checks if the storage contains a given content checksum | """Checks if the storage contains a given content checksum | ||||
Args: query string of the form <hash_algo:hash> | Args: query string of the form <hash_algo:hash> | ||||
Returns: Dict with key found to True or False, according to | Returns: Dict with key found to True or False, according to | ||||
whether the checksum is present or not | whether the checksum is present or not | ||||
""" | """ | ||||
algo, hash = query.parse_hash(q) | algo, hash = query.parse_hash(q) | ||||
found = storage.content_find({algo: hash}) | found = _first_element(storage.content_find({algo: hash})) | ||||
return {'found': found is not None} | return {'found': found is not None} | ||||
def _lookup_content_sha1(q): | def _lookup_content_sha1(q): | ||||
"""Given a possible input, query for the content's sha1. | """Given a possible input, query for the content's sha1. | ||||
Args: | Args: | ||||
q: query string of the form <hash_algo:hash> | q: query string of the form <hash_algo:hash> | ||||
Returns: | Returns: | ||||
binary sha1 if found or None | binary sha1 if found or None | ||||
""" | """ | ||||
algo, hash = query.parse_hash(q) | algo, hash = query.parse_hash(q) | ||||
if algo != 'sha1': | if algo != 'sha1': | ||||
hashes = storage.content_find({algo: hash}) | hashes = _first_element(storage.content_find({algo: hash})) | ||||
if not hashes: | if not hashes: | ||||
return None | return None | ||||
return hashes['sha1'] | return hashes['sha1'] | ||||
return hash | return hash | ||||
def lookup_content_ctags(q): | def lookup_content_ctags(q): | ||||
"""Return ctags information from a specified content. | """Return ctags information from a specified content. | ||||
▲ Show 20 Lines • Show All 554 Lines • ▼ Show 20 Lines | Raises: | ||||
BadInputExc in case of unknown algo_hash or bad hash. | BadInputExc in case of unknown algo_hash or bad hash. | ||||
NotFoundExc either if the revision is not found or the path referenced | NotFoundExc either if the revision is not found or the path referenced | ||||
does not exist. | does not exist. | ||||
NotImplementedError in case of dir_path exists but do not reference a | NotImplementedError in case of dir_path exists but do not reference a | ||||
type 'dir' or 'file'. | type 'dir' or 'file'. | ||||
""" | """ | ||||
sha1_git_bin = _to_sha1_bin(sha1_git) | sha1_git_bin = _to_sha1_bin(sha1_git) | ||||
revision = _first_element(storage.revision_get([sha1_git_bin])) | revision = _first_element(storage.revision_get([sha1_git_bin])) | ||||
if not revision: | if not revision: | ||||
raise NotFoundExc('Revision %s not found' % sha1_git) | raise NotFoundExc('Revision %s not found' % sha1_git) | ||||
dir_sha1_git_bin = revision['directory'] | dir_sha1_git_bin = revision['directory'] | ||||
if dir_path: | if dir_path: | ||||
paths = dir_path.strip(os.path.sep).split(os.path.sep) | paths = dir_path.strip(os.path.sep).split(os.path.sep) | ||||
entity = storage.directory_entry_get_by_path( | entity = storage.directory_entry_get_by_path( | ||||
dir_sha1_git_bin, list(map(lambda p: p.encode('utf-8'), paths))) | dir_sha1_git_bin, list(map(lambda p: p.encode('utf-8'), paths))) | ||||
if not entity: | if not entity: | ||||
raise NotFoundExc( | raise NotFoundExc( | ||||
"Directory or File '%s' pointed to by revision %s not found" | "Directory or File '%s' pointed to by revision %s not found" | ||||
% (dir_path, sha1_git)) | % (dir_path, sha1_git)) | ||||
else: | else: | ||||
entity = {'type': 'dir', 'target': dir_sha1_git_bin} | entity = {'type': 'dir', 'target': dir_sha1_git_bin} | ||||
if entity['type'] == 'dir': | if entity['type'] == 'dir': | ||||
directory_entries = storage.directory_ls(entity['target']) or [] | directory_entries = storage.directory_ls(entity['target']) or [] | ||||
return {'type': 'dir', | return {'type': 'dir', | ||||
'path': '.' if not dir_path else dir_path, | 'path': '.' if not dir_path else dir_path, | ||||
'revision': sha1_git, | 'revision': sha1_git, | ||||
'content': list(map(converters.from_directory_entry, | 'content': list(map(converters.from_directory_entry, | ||||
directory_entries))} | directory_entries))} | ||||
elif entity['type'] == 'file': # content | elif entity['type'] == 'file': # content | ||||
content = storage.content_find({'sha1_git': entity['target']}) | content = _first_element( | ||||
storage.content_find({'sha1_git': entity['target']})) | |||||
if not content: | |||||
anlambert: Use single quotes for string literal | |||||
raise NotFoundExc('Content not found for revision %s' | |||||
Done Inline Actionsthe parenthesis around sha1_git are not needed when there is only a single value to interpolate in the string anlambert: the parenthesis around `sha1_git` are not needed when there is only a single value to… | |||||
% sha1_git) | |||||
if with_data: | if with_data: | ||||
Done Inline ActionsUse only one API call vlorentz: Use only one API call | |||||
Done Inline ActionsJust a nitpick on code style here, can you change it to content = _first_element( storage.content_find({'sha1_git': entity['target']})) anlambert: Just a nitpick on code style here, can you change it to
```lang=python
content = _first_element… | |||||
c = _first_element(storage.content_get([content['sha1']])) | c = _first_element(storage.content_get([content['sha1']])) | ||||
Done Inline ActionsSame kind of formatting issue here, change it to raise NotFoundExc('Content not found for revision %s' % sha1_git) anlambert: Same kind of formatting issue here, change it to
```lang=python
raise NotFoundExc('Content not… | |||||
content['data'] = c['data'] | content['data'] = c['data'] | ||||
Done Inline ActionsThis will crash if content is None. Please fix it and add a test that checks this case. vlorentz: This will crash if `content` is `None`.
Please fix it and add a test that checks this case. | |||||
return {'type': 'file', | return {'type': 'file', | ||||
'path': '.' if not dir_path else dir_path, | 'path': '.' if not dir_path else dir_path, | ||||
'revision': sha1_git, | 'revision': sha1_git, | ||||
'content': converters.from_content(content)} | 'content': converters.from_content(content)} | ||||
elif entity['type'] == 'rev': # revision | elif entity['type'] == 'rev': # revision | ||||
revision = next(storage.revision_get([entity['target']])) | revision = next(storage.revision_get([entity['target']])) | ||||
return {'type': 'rev', | return {'type': 'rev', | ||||
'path': '.' if not dir_path else dir_path, | 'path': '.' if not dir_path else dir_path, | ||||
Show All 10 Lines | def lookup_content(q): | ||||
Args: | Args: | ||||
q: The release's sha1 as hexadecimal | q: The release's sha1 as hexadecimal | ||||
Raises: | Raises: | ||||
NotFoundExc if the requested content is not found | NotFoundExc if the requested content is not found | ||||
""" | """ | ||||
algo, hash = query.parse_hash(q) | algo, hash = query.parse_hash(q) | ||||
c = storage.content_find({algo: hash}) | c = _first_element(storage.content_find({algo: hash})) | ||||
if not c: | if not c: | ||||
raise NotFoundExc('Content with %s checksum equals to %s not found!' % | raise NotFoundExc('Content with %s checksum equals to %s not found!' % | ||||
(algo, hashutil.hash_to_hex(hash))) | (algo, hashutil.hash_to_hex(hash))) | ||||
return converters.from_content(c) | return converters.from_content(c) | ||||
def lookup_content_raw(q): | def lookup_content_raw(q): | ||||
"""Lookup the content defined by q. | """Lookup the content defined by q. | ||||
▲ Show 20 Lines • Show All 298 Lines • Show Last 20 Lines |
Use single quotes for string literal