Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/postgresql/storage.py
| Show First 20 Lines • Show All 1,390 Lines • ▼ Show 20 Lines | |||||
| def metadata_authority_get( | def metadata_authority_get( | ||||
| self, type: MetadataAuthorityType, url: str, db=None, cur=None | self, type: MetadataAuthorityType, url: str, db=None, cur=None | ||||
| ) -> Optional[MetadataAuthority]: | ) -> Optional[MetadataAuthority]: | ||||
| row = db.metadata_authority_get(type.value, url, cur=cur) | row = db.metadata_authority_get(type.value, url, cur=cur) | ||||
| if not row: | if not row: | ||||
| return None | return None | ||||
| return MetadataAuthority.from_dict(dict(zip(db.metadata_authority_cols, row))) | return MetadataAuthority.from_dict(dict(zip(db.metadata_authority_cols, row))) | ||||
| def clear_buffers(self, object_types: Sequence[str]) -> None: | def clear_buffers(self, object_types: Sequence[str] = ()) -> None: | ||||
| """Do nothing | """Do nothing | ||||
| """ | """ | ||||
| return None | return None | ||||
| def flush(self, object_types: Sequence[str]) -> Dict[str, int]: | def flush(self, object_types: Sequence[str] = ()) -> Dict[str, int]: | ||||
| return {} | return {} | ||||
| def _get_authority_id(self, authority: MetadataAuthority, db, cur): | def _get_authority_id(self, authority: MetadataAuthority, db, cur): | ||||
| authority_id = db.metadata_authority_get_id( | authority_id = db.metadata_authority_get_id( | ||||
| authority.type.value, authority.url, cur | authority.type.value, authority.url, cur | ||||
| ) | ) | ||||
| if not authority_id: | if not authority_id: | ||||
| raise StorageArgumentException(f"Unknown authority {authority}") | raise StorageArgumentException(f"Unknown authority {authority}") | ||||
| return authority_id | return authority_id | ||||
| def _get_fetcher_id(self, fetcher: MetadataFetcher, db, cur): | def _get_fetcher_id(self, fetcher: MetadataFetcher, db, cur): | ||||
| fetcher_id = db.metadata_fetcher_get_id(fetcher.name, fetcher.version, cur) | fetcher_id = db.metadata_fetcher_get_id(fetcher.name, fetcher.version, cur) | ||||
| if not fetcher_id: | if not fetcher_id: | ||||
| raise StorageArgumentException(f"Unknown fetcher {fetcher}") | raise StorageArgumentException(f"Unknown fetcher {fetcher}") | ||||
| return fetcher_id | return fetcher_id | ||||