Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/cassandra/storage.py
| Show First 20 Lines • Show All 1,022 Lines • ▼ Show 20 Lines | def origin_metadata_add( | ||||
| discovery_date: datetime.datetime, | discovery_date: datetime.datetime, | ||||
| authority: Dict[str, Any], | authority: Dict[str, Any], | ||||
| fetcher: Dict[str, Any], | fetcher: Dict[str, Any], | ||||
| format: str, | format: str, | ||||
| metadata: bytes, | metadata: bytes, | ||||
| ) -> None: | ) -> None: | ||||
| if not isinstance(origin_url, str): | if not isinstance(origin_url, str): | ||||
| raise StorageArgumentException( | raise StorageArgumentException( | ||||
| "origin_id must be str, not %r" % (origin_url,) | "origin_url must be str, not %r" % (origin_url,) | ||||
| ) | ) | ||||
| self._object_metadata_add( | |||||
| "origin", origin_url, discovery_date, authority, fetcher, format, metadata, | |||||
| ) | |||||
| def origin_metadata_get( | |||||
| self, | |||||
| origin_url: str, | |||||
| authority: Dict[str, str], | |||||
| after: Optional[datetime.datetime] = None, | |||||
| page_token: Optional[bytes] = None, | |||||
| limit: int = 1000, | |||||
| ) -> Dict[str, Any]: | |||||
| if not isinstance(origin_url, str): | |||||
| raise TypeError("origin_url must be str, not %r" % (origin_url,)) | |||||
| res = self._object_metadata_get( | |||||
| "origin", origin_url, authority, after, page_token, limit | |||||
| ) | |||||
| for result in res["results"]: | |||||
| result["origin_url"] = result.pop("id") | |||||
| return res | |||||
| def _object_metadata_add( | |||||
| self, | |||||
| object_type: str, | |||||
| id: str, | |||||
| discovery_date: datetime.datetime, | |||||
| authority: Dict[str, Any], | |||||
| fetcher: Dict[str, Any], | |||||
| format: str, | |||||
| metadata: bytes, | |||||
| ) -> None: | |||||
| if not self._cql_runner.metadata_authority_get(**authority): | if not self._cql_runner.metadata_authority_get(**authority): | ||||
| raise StorageArgumentException(f"Unknown authority {authority}") | raise StorageArgumentException(f"Unknown authority {authority}") | ||||
| if not self._cql_runner.metadata_fetcher_get(**fetcher): | if not self._cql_runner.metadata_fetcher_get(**fetcher): | ||||
| raise StorageArgumentException(f"Unknown fetcher {fetcher}") | raise StorageArgumentException(f"Unknown fetcher {fetcher}") | ||||
| try: | try: | ||||
| self._cql_runner.origin_metadata_add( | self._cql_runner.object_metadata_add( | ||||
| origin_url, | object_type, | ||||
| id, | |||||
| authority["type"], | authority["type"], | ||||
| authority["url"], | authority["url"], | ||||
| discovery_date, | discovery_date, | ||||
| fetcher["name"], | fetcher["name"], | ||||
| fetcher["version"], | fetcher["version"], | ||||
| format, | format, | ||||
| metadata, | metadata, | ||||
| ) | ) | ||||
| except TypeError as e: | except TypeError as e: | ||||
| raise StorageArgumentException(*e.args) | raise StorageArgumentException(*e.args) | ||||
| def origin_metadata_get( | def _object_metadata_get( | ||||
| self, | self, | ||||
| origin_url: str, | object_type: str, | ||||
| id: str, | |||||
| authority: Dict[str, str], | authority: Dict[str, str], | ||||
| after: Optional[datetime.datetime] = None, | after: Optional[datetime.datetime] = None, | ||||
| page_token: Optional[bytes] = None, | page_token: Optional[bytes] = None, | ||||
| limit: int = 1000, | limit: int = 1000, | ||||
| ) -> Dict[str, Any]: | ) -> Dict[str, Any]: | ||||
| if not isinstance(origin_url, str): | |||||
| raise TypeError("origin_url must be str, not %r" % (origin_url,)) | |||||
| if page_token is not None: | if page_token is not None: | ||||
| (after_date, after_fetcher_name, after_fetcher_url) = msgpack_loads( | (after_date, after_fetcher_name, after_fetcher_url) = msgpack_loads( | ||||
| page_token | page_token | ||||
| ) | ) | ||||
| if after and after_date < after: | if after and after_date < after: | ||||
| raise StorageArgumentException( | raise StorageArgumentException( | ||||
| "page_token is inconsistent with the value of 'after'." | "page_token is inconsistent with the value of 'after'." | ||||
| ) | ) | ||||
| entries = self._cql_runner.origin_metadata_get_after_date_and_fetcher( | entries = self._cql_runner.object_metadata_get_after_date_and_fetcher( | ||||
| origin_url, | id, | ||||
| authority["type"], | authority["type"], | ||||
| authority["url"], | authority["url"], | ||||
| after_date, | after_date, | ||||
| after_fetcher_name, | after_fetcher_name, | ||||
| after_fetcher_url, | after_fetcher_url, | ||||
| ) | ) | ||||
| elif after is not None: | elif after is not None: | ||||
| entries = self._cql_runner.origin_metadata_get_after_date( | entries = self._cql_runner.object_metadata_get_after_date( | ||||
| origin_url, authority["type"], authority["url"], after | id, authority["type"], authority["url"], after | ||||
| ) | ) | ||||
| else: | else: | ||||
| entries = self._cql_runner.origin_metadata_get( | entries = self._cql_runner.object_metadata_get( | ||||
| origin_url, authority["type"], authority["url"] | id, authority["type"], authority["url"] | ||||
| ) | ) | ||||
| if limit: | if limit: | ||||
| entries = itertools.islice(entries, 0, limit + 1) | entries = itertools.islice(entries, 0, limit + 1) | ||||
| results = [] | results = [] | ||||
| for entry in entries: | for entry in entries: | ||||
| discovery_date = entry.discovery_date.replace(tzinfo=datetime.timezone.utc) | discovery_date = entry.discovery_date.replace(tzinfo=datetime.timezone.utc) | ||||
| results.append( | |||||
| { | result = { | ||||
| "origin_url": entry.origin, | "id": entry.id, | ||||
| "authority": { | "authority": { | ||||
| "type": entry.authority_type, | "type": entry.authority_type, | ||||
| "url": entry.authority_url, | "url": entry.authority_url, | ||||
| }, | }, | ||||
| "fetcher": { | "fetcher": { | ||||
| "name": entry.fetcher_name, | "name": entry.fetcher_name, | ||||
| "version": entry.fetcher_version, | "version": entry.fetcher_version, | ||||
| }, | }, | ||||
| "discovery_date": discovery_date, | "discovery_date": discovery_date, | ||||
| "format": entry.format, | "format": entry.format, | ||||
| "metadata": entry.metadata, | "metadata": entry.metadata, | ||||
| } | } | ||||
| ) | |||||
| results.append(result) | |||||
| if len(results) > limit: | if len(results) > limit: | ||||
| results.pop() | results.pop() | ||||
| assert len(results) == limit | assert len(results) == limit | ||||
| last_result = results[-1] | last_result = results[-1] | ||||
| next_page_token: Optional[bytes] = msgpack_dumps( | next_page_token: Optional[bytes] = msgpack_dumps( | ||||
| ( | ( | ||||
| last_result["discovery_date"], | last_result["discovery_date"], | ||||
| ▲ Show 20 Lines • Show All 52 Lines • Show Last 20 Lines | |||||