Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/in_memory.py
| Show First 20 Lines • Show All 684 Lines • ▼ Show 20 Lines | def origin_get_range(self, origin_from=1, origin_count=100): | ||||
| if origin_from <= len(self._origins_by_id): | if origin_from <= len(self._origins_by_id): | ||||
| max_idx = origin_from + origin_count - 1 | max_idx = origin_from + origin_count - 1 | ||||
| if max_idx > len(self._origins_by_id): | if max_idx > len(self._origins_by_id): | ||||
| max_idx = len(self._origins_by_id) | max_idx = len(self._origins_by_id) | ||||
| for idx in range(origin_from - 1, max_idx): | for idx in range(origin_from - 1, max_idx): | ||||
| origin = self._convert_origin(self._origins[self._origins_by_id[idx]]) | origin = self._convert_origin(self._origins[self._origins_by_id[idx]]) | ||||
| yield {"id": idx + 1, **origin} | yield {"id": idx + 1, **origin} | ||||
| def origin_list(self, page_token: Optional[str] = None, limit: int = 100) -> dict: | def origin_list( | ||||
| self, page_token: Optional[str] = None, limit: int = 100 | |||||
| ) -> PagedResult[Origin]: | |||||
| origin_urls = sorted(self._origins) | origin_urls = sorted(self._origins) | ||||
| if page_token: | from_ = bisect.bisect_left(origin_urls, page_token) if page_token else 0 | ||||
| from_ = bisect.bisect_left(origin_urls, page_token) | next_page_token = None | ||||
| else: | |||||
| from_ = 0 | |||||
| result = { | # Take one more origin so we can reuse it as the next page token if any | ||||
| "origins": [ | origins = [Origin(url=url) for url in origin_urls[from_ : from_ + limit + 1]] | ||||
| {"url": origin_url} for origin_url in origin_urls[from_ : from_ + limit] | |||||
| ] | |||||
| } | |||||
| if from_ + limit < len(origin_urls): | if len(origins) > limit: | ||||
| result["next_page_token"] = origin_urls[from_ + limit] | # last origin id is the next page token | ||||
| next_page_token = str(origins[-1].url) | |||||
| # excluding that origin from the result to respect the limit size | |||||
| origins = origins[:limit] | |||||
| return result | assert len(origins) <= limit | ||||
| return PagedResult(results=origins, next_page_token=next_page_token) | |||||
| def origin_search( | def origin_search( | ||||
| self, url_pattern, offset=0, limit=50, regexp=False, with_visit=False | self, url_pattern, offset=0, limit=50, regexp=False, with_visit=False | ||||
| ): | ): | ||||
| origins = map(self._convert_origin, self._origins.values()) | origins = map(self._convert_origin, self._origins.values()) | ||||
| if regexp: | if regexp: | ||||
| pat = re.compile(url_pattern) | pat = re.compile(url_pattern) | ||||
| origins = [orig for orig in origins if pat.search(orig["url"])] | origins = [orig for orig in origins if pat.search(orig["url"])] | ||||
| ▲ Show 20 Lines • Show All 537 Lines • Show Last 20 Lines | |||||