Changeset View
Changeset View
Standalone View
Standalone View
swh/web/api/utils.py
# Copyright (C) 2015-2019 The Software Heritage developers | # Copyright (C) 2015-2019 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU Affero General Public License version 3, or any later version | # License: GNU Affero General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
from typing import Dict, Optional, Any | |||||
from swh.web.common.utils import reverse | from django.http import HttpRequest | ||||
from swh.web.common.utils import reverse, resolve_branch_alias | |||||
from swh.web.common.query import parse_hash | from swh.web.common.query import parse_hash | ||||
def filter_field_keys(data, field_keys): | def filter_field_keys(data, field_keys): | ||||
"""Given an object instance (directory or list), and a csv field keys | """Given an object instance (directory or list), and a csv field keys | ||||
to filter on. | to filter on. | ||||
Return the object instance with filtered keys. | Return the object instance with filtered keys. | ||||
Show All 20 Lines | |||||
def person_to_string(person): | def person_to_string(person): | ||||
"""Map a person (person, committer, tagger, etc...) to a string. | """Map a person (person, committer, tagger, etc...) to a string. | ||||
""" | """ | ||||
return ''.join([person['name'], ' <', person['email'], '>']) | return ''.join([person['name'], ' <', person['email'], '>']) | ||||
def enrich_object(object): | def enrich_object(object: Dict[str, str], | ||||
request: Optional[HttpRequest] = None) -> Dict[str, str]: | |||||
"""Enrich an object (revision, release) with link to the 'target' of | """Enrich an object (revision, release) with link to the 'target' of | ||||
type 'target_type'. | type 'target_type'. | ||||
Args: | Args: | ||||
object: An object with target and target_type keys | object: An object with target and target_type keys | ||||
(e.g. release, revision) | (e.g. release, revision) | ||||
request: Absolute URIs will be generated if provided | |||||
Returns: | Returns: | ||||
Object enriched with target_url pointing to the right | Object enriched with target object url (revision, release, content, | ||||
swh.web.ui.api urls for the pointing object (revision, | directory) | ||||
release, content, directory) | |||||
""" | |||||
""" | if 'target' in object and 'target_type' in object: | ||||
obj = object.copy() | if object['target_type'] in ('revision', 'release', 'directory'): | ||||
if 'target' in obj and 'target_type' in obj: | object['target_url'] = reverse( | ||||
if obj['target_type'] in ('revision', 'release', 'directory'): | 'api-1-%s' % object['target_type'], | ||||
obj['target_url'] = \ | url_args={'sha1_git': object['target']}, | ||||
reverse('api-1-%s' % obj['target_type'], | request=request) | ||||
url_args={'sha1_git': obj['target']}) | elif object['target_type'] == 'content': | ||||
elif obj['target_type'] == 'content': | object['target_url'] = reverse( | ||||
obj['target_url'] = \ | 'api-1-content', | ||||
reverse('api-1-content', | url_args={'q': 'sha1_git:' + object['target']}, | ||||
url_args={'q': 'sha1_git:' + obj['target']}) | request=request) | ||||
elif obj['target_type'] == 'snapshot': | elif object['target_type'] == 'snapshot': | ||||
obj['target_url'] = \ | object['target_url'] = reverse( | ||||
reverse('api-1-snapshot', | 'api-1-snapshot', | ||||
url_args={'snapshot_id': obj['target']}) | url_args={'snapshot_id': object['target']}, | ||||
request=request) | |||||
return obj | return object | ||||
enrich_release = enrich_object | enrich_release = enrich_object | ||||
def enrich_directory(directory, context_url=None): | def enrich_directory(directory: Dict[str, str], | ||||
request: Optional[HttpRequest] = None) -> Dict[str, str]: | |||||
"""Enrich directory with url to content or directory. | """Enrich directory with url to content or directory. | ||||
Args: | |||||
directory: dict of data associated to a swh directory object | |||||
request: Absolute URIs will be generated if provided | |||||
Returns: | |||||
An enriched directory dict filled with additional urls | |||||
""" | """ | ||||
if 'type' in directory: | if 'type' in directory: | ||||
target_type = directory['type'] | target_type = directory['type'] | ||||
target = directory['target'] | target = directory['target'] | ||||
if target_type == 'file': | if target_type == 'file': | ||||
directory['target_url'] = reverse( | directory['target_url'] = reverse( | ||||
'api-1-content', url_args={'q': 'sha1_git:%s' % target}) | 'api-1-content', | ||||
if context_url: | url_args={'q': 'sha1_git:%s' % target}, | ||||
directory['file_url'] = context_url + directory['name'] + '/' | request=request) | ||||
elif target_type == 'dir': | elif target_type == 'dir': | ||||
directory['target_url'] = reverse( | directory['target_url'] = reverse( | ||||
'api-1-directory', url_args={'sha1_git': target}) | 'api-1-directory', | ||||
if context_url: | url_args={'sha1_git': target}, | ||||
directory['dir_url'] = context_url + directory['name'] + '/' | request=request) | ||||
else: | else: | ||||
directory['target_url'] = reverse( | directory['target_url'] = reverse( | ||||
'api-1-revision', url_args={'sha1_git': target}) | 'api-1-revision', | ||||
if context_url: | url_args={'sha1_git': target}, | ||||
directory['rev_url'] = context_url + directory['name'] + '/' | request=request) | ||||
return directory | return directory | ||||
def enrich_metadata_endpoint(content): | def enrich_metadata_endpoint(content_metadata: Dict[str, str], | ||||
"""Enrich metadata endpoint with link to the upper metadata endpoint. | request: Optional[HttpRequest] = None | ||||
) -> Dict[str, str]: | |||||
"""Enrich content metadata dict with link to the upper metadata endpoint. | |||||
Args: | |||||
content_metadata: dict of data associated to a swh content metadata | |||||
request: Absolute URIs will be generated if provided | |||||
Returns: | |||||
An enriched content metadata dict filled with an additional url | |||||
""" | """ | ||||
c = content.copy() | c = content_metadata | ||||
c['content_url'] = reverse('api-1-content', | c['content_url'] = reverse('api-1-content', | ||||
url_args={'q': 'sha1:%s' % c['id']}) | url_args={'q': 'sha1:%s' % c['id']}, | ||||
request=request) | |||||
return c | return c | ||||
def enrich_content(content, top_url=False, query_string=None): | def enrich_content(content: Dict[str, Any], | ||||
top_url: Optional[bool] = False, | |||||
query_string: Optional[str] = None, | |||||
request: Optional[HttpRequest] = None) -> Dict[str, str]: | |||||
"""Enrich content with links to: | """Enrich content with links to: | ||||
- data_url: its raw data | - data_url: its raw data | ||||
- filetype_url: its filetype information | - filetype_url: its filetype information | ||||
- language_url: its programming language information | - language_url: its programming language information | ||||
- license_url: its licensing information | - license_url: its licensing information | ||||
Args: | Args: | ||||
content: dict of data associated to a swh content object | content: dict of data associated to a swh content object | ||||
top_url: whether or not to include the content url in | top_url: whether or not to include the content url in | ||||
the enriched data | the enriched data | ||||
query_string: optional query string of type '<algo>:<hash>' | query_string: optional query string of type '<algo>:<hash>' | ||||
used when requesting the content, it acts as a hint | used when requesting the content, it acts as a hint | ||||
for picking the same hash method when computing | for picking the same hash method when computing | ||||
the url listed above | the url listed above | ||||
request: Absolute URIs will be generated if provided | |||||
Returns: | Returns: | ||||
An enriched content dict filled with additional urls | An enriched content dict filled with additional urls | ||||
""" | """ | ||||
checksums = content | checksums = content | ||||
if 'checksums' in content: | if 'checksums' in content: | ||||
checksums = content['checksums'] | checksums = content['checksums'] | ||||
hash_algo = 'sha1' | hash_algo = 'sha1' | ||||
if query_string: | if query_string: | ||||
hash_algo = parse_hash(query_string)[0] | hash_algo = parse_hash(query_string)[0] | ||||
if hash_algo in checksums: | if hash_algo in checksums: | ||||
q = '%s:%s' % (hash_algo, checksums[hash_algo]) | q = '%s:%s' % (hash_algo, checksums[hash_algo]) | ||||
if top_url: | if top_url: | ||||
content['content_url'] = reverse( | content['content_url'] = reverse( | ||||
'api-1-content', url_args={'q': q}) | 'api-1-content', url_args={'q': q}) | ||||
content['data_url'] = reverse('api-1-content-raw', url_args={'q': q}) | content['data_url'] = reverse('api-1-content-raw', | ||||
url_args={'q': q}, | |||||
request=request) | |||||
content['filetype_url'] = reverse( | content['filetype_url'] = reverse( | ||||
'api-1-content-filetype', url_args={'q': q}) | 'api-1-content-filetype', | ||||
url_args={'q': q}, | |||||
request=request) | |||||
content['language_url'] = reverse( | content['language_url'] = reverse( | ||||
'api-1-content-language', url_args={'q': q}) | 'api-1-content-language', | ||||
url_args={'q': q}, | |||||
request=request) | |||||
content['license_url'] = reverse( | content['license_url'] = reverse( | ||||
'api-1-content-license', url_args={'q': q}) | 'api-1-content-license', | ||||
url_args={'q': q}, | |||||
request=request) | |||||
return content | return content | ||||
def enrich_revision(revision): | def enrich_revision(revision: Dict[str, Any], | ||||
request: Optional[HttpRequest] = None) -> Dict[str, Any]: | |||||
"""Enrich revision with links where it makes sense (directory, parents). | """Enrich revision with links where it makes sense (directory, parents). | ||||
Keep track of the navigation breadcrumbs if they are specified. | Keep track of the navigation breadcrumbs if they are specified. | ||||
Args: | Args: | ||||
revision: the revision as a dict | revision: the revision as a dict | ||||
request: Absolute URIs will be generated if provided | |||||
Returns: | |||||
An enriched revision dict filled with additional urls | |||||
""" | """ | ||||
revision['url'] = reverse('api-1-revision', | revision['url'] = reverse('api-1-revision', | ||||
url_args={'sha1_git': revision['id']}) | url_args={'sha1_git': revision['id']}, | ||||
request=request) | |||||
revision['history_url'] = reverse('api-1-revision-log', | revision['history_url'] = reverse('api-1-revision-log', | ||||
url_args={'sha1_git': revision['id']}) | url_args={'sha1_git': revision['id']}, | ||||
request=request) | |||||
if 'directory' in revision: | if 'directory' in revision: | ||||
revision['directory_url'] = reverse( | revision['directory_url'] = reverse( | ||||
'api-1-directory', url_args={'sha1_git': revision['directory']}) | 'api-1-directory', | ||||
url_args={'sha1_git': revision['directory']}, | |||||
request=request) | |||||
if 'parents' in revision: | if 'parents' in revision: | ||||
parents = [] | parents = [] | ||||
for parent in revision['parents']: | for parent in revision['parents']: | ||||
parents.append({ | parents.append({ | ||||
'id': parent, | 'id': parent, | ||||
'url': reverse('api-1-revision', url_args={'sha1_git': parent}) | 'url': reverse('api-1-revision', | ||||
url_args={'sha1_git': parent}, | |||||
request=request) | |||||
}) | }) | ||||
revision['parents'] = parents | revision['parents'] = parents | ||||
if 'children' in revision: | if 'children' in revision: | ||||
children = [] | children = [] | ||||
for child in revision['children']: | for child in revision['children']: | ||||
children.append(reverse( | children.append(reverse( | ||||
'api-1-revision', url_args={'sha1_git': child})) | 'api-1-revision', | ||||
url_args={'sha1_git': child}, | |||||
request=request)) | |||||
revision['children_urls'] = children | revision['children_urls'] = children | ||||
if 'message_decoding_failed' in revision: | if 'message_decoding_failed' in revision: | ||||
revision['message_url'] = \ | revision['message_url'] = reverse( | ||||
reverse('api-1-revision-raw-message', | 'api-1-revision-raw-message', | ||||
url_args={'sha1_git': revision['id']}) | url_args={'sha1_git': revision['id']}, | ||||
request=request) | |||||
return revision | return revision | ||||
def enrich_snapshot(snapshot: Dict[str, Any], | |||||
request: Optional[HttpRequest] = None) -> Dict[str, Any]: | |||||
"""Enrich snapshot with links to the branch targets | |||||
Args: | |||||
snapshot: the snapshot as a dict | |||||
request: Absolute URIs will be generated if provided | |||||
Returns: | |||||
An enriched snapshot dict filled with additional urls | |||||
""" | |||||
if 'branches' in snapshot: | |||||
snapshot['branches'] = { | |||||
k: enrich_object(v, request) if v else None | |||||
for k, v in snapshot['branches'].items() | |||||
} | |||||
for k, v in snapshot['branches'].items(): | |||||
if v and v['target_type'] == 'alias': | |||||
branch = resolve_branch_alias(snapshot, v) | |||||
if branch: | |||||
branch = enrich_object(branch, request) | |||||
v['target_url'] = branch['target_url'] | |||||
return snapshot | |||||
def enrich_origin(origin: Dict[str, Any], | |||||
request: Optional[HttpRequest] = None) -> Dict[str, Any]: | |||||
"""Enrich origin dict with link to its visits | |||||
Args: | |||||
origin: the origin as a dict | |||||
request: Absolute URIs will be generated if provided | |||||
Returns: | |||||
An enriched origin dict filled with an additional url | |||||
""" | |||||
if 'url' in origin: | |||||
origin['origin_visits_url'] = reverse( | |||||
'api-1-origin-visits', | |||||
url_args={'origin_url': origin['url']}, | |||||
request=request) | |||||
return origin | |||||
def enrich_origin_visit(origin_visit: Dict[str, Any], *, | |||||
with_origin_link: bool, with_origin_visit_link: bool, | |||||
request: Optional[HttpRequest] = None | |||||
) -> Dict[str, Any]: | |||||
"""Enrich origin visit dict with additional links | |||||
Args: | |||||
origin_visit: the origin visit as a dict | |||||
with_origin_link: whether to add link to origin | |||||
with_origin_visit_link: whether to add link to origin visit | |||||
request: Absolute URIs will be generated if provided | |||||
Returns: | |||||
An enriched origin visit dict filled with additional urls | |||||
""" | |||||
ov = origin_visit | |||||
if with_origin_link: | |||||
ov['origin_url'] = reverse('api-1-origin', | |||||
url_args={'origin_url': ov['origin']}, | |||||
request=request) | |||||
if with_origin_visit_link: | |||||
ov['origin_visit_url'] = reverse('api-1-origin-visit', | |||||
url_args={'origin_url': ov['origin'], | |||||
'visit_id': ov['visit']}, | |||||
request=request) | |||||
snapshot = ov['snapshot'] | |||||
if snapshot: | |||||
ov['snapshot_url'] = reverse('api-1-snapshot', | |||||
url_args={'snapshot_id': snapshot}, | |||||
request=request) | |||||
else: | |||||
ov['snapshot_url'] = None | |||||
return ov |