Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/in_memory.py
Show First 20 Lines • Show All 698 Lines • ▼ Show 20 Lines | def _convert_origin(self, t): | ||||
return t.to_dict() | return t.to_dict() | ||||
def origin_get_one(self, origin_url: str) -> Optional[Origin]: | def origin_get_one(self, origin_url: str) -> Optional[Origin]: | ||||
return self._origins.get(origin_url) | return self._origins.get(origin_url) | ||||
def origin_get(self, origins: List[str]) -> Iterable[Optional[Origin]]: | def origin_get(self, origins: List[str]) -> Iterable[Optional[Origin]]: | ||||
return [self.origin_get_one(origin_url) for origin_url in origins] | return [self.origin_get_one(origin_url) for origin_url in origins] | ||||
def origin_get_by_sha1( | def origin_get_by_sha1(self, sha1s: List[bytes]) -> List[Optional[Dict[str, Any]]]: | ||||
self, sha1s: List[bytes] | return [self._convert_origin(self._origins_by_sha1.get(sha1)) for sha1 in sha1s] | ||||
) -> Iterable[Optional[Dict[str, Any]]]: | |||||
for sha1 in sha1s: | |||||
yield self._convert_origin(self._origins_by_sha1.get(sha1)) | |||||
def origin_list( | def origin_list( | ||||
self, page_token: Optional[str] = None, limit: int = 100 | self, page_token: Optional[str] = None, limit: int = 100 | ||||
) -> PagedResult[Origin]: | ) -> PagedResult[Origin]: | ||||
origin_urls = sorted(self._origins) | origin_urls = sorted(self._origins) | ||||
from_ = bisect.bisect_left(origin_urls, page_token) if page_token else 0 | from_ = bisect.bisect_left(origin_urls, page_token) if page_token else 0 | ||||
next_page_token = None | next_page_token = None | ||||
▲ Show 20 Lines • Show All 579 Lines • Show Last 20 Lines |