Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/interface.py
| Show First 20 Lines • Show All 49 Lines • ▼ Show 20 Lines | class PartialBranches(TypedDict): | ||||
| branches: Dict[bytes, Optional[SnapshotBranch]] | branches: Dict[bytes, Optional[SnapshotBranch]] | ||||
| """A dict of branches contained in the snapshot | """A dict of branches contained in the snapshot | ||||
| whose keys are the branches' names""" | whose keys are the branches' names""" | ||||
| next_branch: Optional[bytes] | next_branch: Optional[bytes] | ||||
| """The name of the first branch not returned or :const:`None` if | """The name of the first branch not returned or :const:`None` if | ||||
| the snapshot has less than the request number of branches.""" | the snapshot has less than the request number of branches.""" | ||||
| class HashDict(TypedDict, total=False): | |||||
| sha1: bytes | |||||
| sha1_git: bytes | |||||
| sha256: bytes | |||||
| blake2s256: bytes | |||||
| class TotalHashDict(HashDict, total=True): | |||||
| pass | |||||
| @attr.s | @attr.s | ||||
| class OriginVisitWithStatuses: | class OriginVisitWithStatuses: | ||||
| visit = attr.ib(type=OriginVisit) | visit = attr.ib(type=OriginVisit) | ||||
| statuses = attr.ib(type=List[OriginVisitStatus]) | statuses = attr.ib(type=List[OriginVisitStatus]) | ||||
| TResult = TypeVar("TResult") | TResult = TypeVar("TResult") | ||||
| PagedResult = CorePagedResult[TResult, str] | PagedResult = CorePagedResult[TResult, str] | ||||
| ▲ Show 20 Lines • Show All 157 Lines • ▼ Show 20 Lines | ) -> List[Optional[Content]]: | ||||
| Returns: | Returns: | ||||
| List of contents model objects when they exist, None otherwise. | List of contents model objects when they exist, None otherwise. | ||||
| """ | """ | ||||
| ... | ... | ||||
| @remote_api_endpoint("content/missing") | @remote_api_endpoint("content/missing") | ||||
| def content_missing( | def content_missing( | ||||
| self, contents: List[Dict[str, Any]], key_hash: str = "sha1" | self, contents: List[HashDict], key_hash: str = "sha1" | ||||
| ) -> Iterable[bytes]: | ) -> Iterable[bytes]: | ||||
| """List content missing from storage | """List content missing from storage | ||||
| Args: | Args: | ||||
| content: iterable of dictionaries whose keys are either 'length' or an item | content: iterable of dictionaries whose keys are either 'length' or an item | ||||
| of :data:`swh.model.hashutil.ALGORITHMS`; mapped to the | of :data:`swh.model.hashutil.ALGORITHMS`; mapped to the | ||||
| corresponding checksum (or length). | corresponding checksum (or length). | ||||
| key_hash: name of the column to use as hash id result (default: 'sha1') | key_hash: name of the column to use as hash id result (default: 'sha1') | ||||
| Show All 35 Lines | ) -> Iterable[Sha1Git]: | ||||
| Yields: | Yields: | ||||
| missing contents sha1_git | missing contents sha1_git | ||||
| """ | """ | ||||
| ... | ... | ||||
| @remote_api_endpoint("content/present") | @remote_api_endpoint("content/present") | ||||
| def content_find(self, content: Dict[str, Any]) -> List[Content]: | def content_find(self, content: HashDict) -> List[Content]: | ||||
| """Find a content hash in db. | """Find a content hash in db. | ||||
| Args: | Args: | ||||
| content: a dictionary representing one content hash, mapping | content: a dictionary representing one content hash, mapping | ||||
| checksum algorithm names (see swh.model.hashutil.ALGORITHMS) to | checksum algorithm names (see swh.model.hashutil.ALGORITHMS) to | ||||
| checksum values | checksum values | ||||
| Raises: | Raises: | ||||
| ▲ Show 20 Lines • Show All 1,115 Lines • Show Last 20 Lines | |||||