Changeset View
Changeset View
Standalone View
Standalone View
swh/web/common/archive.py
| Show First 20 Lines • Show All 1,366 Lines • ▼ Show 20 Lines | Args: | ||||
| constructor | constructor | ||||
| """ | """ | ||||
| # first check if the provided revision is valid | # first check if the provided revision is valid | ||||
| lookup_revision(rev_start) | lookup_revision(rev_start) | ||||
| return _RevisionsWalkerProxy(rev_walker_type, rev_start, *args, **kwargs) | return _RevisionsWalkerProxy(rev_walker_type, rev_start, *args, **kwargs) | ||||
| def lookup_object(object_type: ObjectType, object_id: str) -> Dict[str, Any]: | def lookup_object( | ||||
| object_type: ObjectType, object_id: str, *, complete: bool = False | |||||
| ) -> Dict[str, Any]: | |||||
| """ | """ | ||||
| Utility function for looking up an object in the archive by its type | Utility function for looking up an object in the archive by its type | ||||
| and id. | and id. | ||||
| Args: | Args: | ||||
| object_type (str): the type of object to lookup, either *content*, | object_type (str): the type of object to lookup, either *content*, | ||||
| *directory*, *release*, *revision* or *snapshot* | *directory*, *release*, *revision* or *snapshot* | ||||
| object_id (str): the *sha1_git* checksum identifier in hexadecimal | object_id (str): the *sha1_git* checksum identifier in hexadecimal | ||||
| form of the object to lookup | form of the object to lookup | ||||
| complete (bool): Whether to include full objects, rather than their | |||||
| default truncations. | |||||
| Currently just affects snapshots. | |||||
| Returns: | Returns: | ||||
| Dict[str, Any]: A dictionary describing the object or a list of | Dict[str, Any]: A dictionary describing the object or a list of | ||||
| dictionary for the directory object type. | dictionary for the directory object type. | ||||
| Raises: | Raises: | ||||
| swh.web.common.exc.NotFoundExc: if the object could not be found in | swh.web.common.exc.NotFoundExc: if the object could not be found in | ||||
| the archive | the archive | ||||
| BadInputExc: if the object identifier is invalid | BadInputExc: if the object identifier is invalid | ||||
| """ | """ | ||||
| if object_type == ObjectType.CONTENT: | if object_type == ObjectType.CONTENT: | ||||
| return lookup_content(f"sha1_git:{object_id}") | return lookup_content(f"sha1_git:{object_id}") | ||||
| elif object_type == ObjectType.DIRECTORY: | elif object_type == ObjectType.DIRECTORY: | ||||
| return {"id": object_id, "content": list(lookup_directory(object_id))} | return {"id": object_id, "content": list(lookup_directory(object_id))} | ||||
| elif object_type == ObjectType.RELEASE: | elif object_type == ObjectType.RELEASE: | ||||
| return lookup_release(object_id) | return lookup_release(object_id) | ||||
| elif object_type == ObjectType.REVISION: | elif object_type == ObjectType.REVISION: | ||||
| return lookup_revision(object_id) | return lookup_revision(object_id) | ||||
| elif object_type == ObjectType.SNAPSHOT: | elif object_type == ObjectType.SNAPSHOT: | ||||
| if complete: | |||||
| return lookup_snapshot(object_id, branch_name_exclude_prefix=None) | |||||
| else: | |||||
| return lookup_snapshot(object_id) | return lookup_snapshot(object_id) | ||||
| else: | else: | ||||
| raise ValueError(f"Unexpected object type variant: {object_type}") | raise ValueError(f"Unexpected object type variant: {object_type}") | ||||
| def lookup_missing_hashes(grouped_swhids: Dict[str, List[bytes]]) -> Set[str]: | def lookup_missing_hashes(grouped_swhids: Dict[str, List[bytes]]) -> Set[str]: | ||||
| """Lookup missing Software Heritage persistent identifier hash, using | """Lookup missing Software Heritage persistent identifier hash, using | ||||
| batch processing. | batch processing. | ||||
| ▲ Show 20 Lines • Show All 41 Lines • Show Last 20 Lines | |||||