swh.storage.postgresql.storage module

swh.storage.postgresql.storage.EMPTY_SNAPSHOT_ID = b'\x1a\x88\x93\xe6\xa8oDN\x8b\xe8\xe7\xbd\xa6\xcb4\xfb\x175\xa0\x0e'

Identifier for the empty snapshot

swh.storage.postgresql.storage.VALIDATION_EXCEPTIONS = (<class 'KeyError'>, <class 'TypeError'>, <class 'ValueError'>, <class 'psycopg2.errors.CheckViolation'>, <class 'psycopg2.IntegrityError'>, <class 'psycopg2.errors.InvalidTextRepresentation'>, <class 'psycopg2.errors.NotNullViolation'>, <class 'psycopg2.errors.NumericValueOutOfRange'>, <class 'psycopg2.errors.UndefinedFunction'>, <class 'psycopg2.errors.ProgramLimitExceeded'>)

Exceptions raised by postgresql when validation of the arguments failed.

swh.storage.postgresql.storage.convert_validation_exceptions()

Catches postgresql errors related to invalid arguments, and re-raises a StorageArgumentException.

swh.storage.postgresql.storage.db_transaction_generator(*args, **kwargs)

swh.storage.postgresql.storage.db_transaction(*args, **kwargs)

class swh.storage.postgresql.storage.Storage(db: str | Db, objstorage: Dict | None = None, min_pool_conns: int = 1, max_pool_conns: int = 10, journal_writer: Dict[str, Any] | None = None, query_options: Dict[str, Dict[str, Any]] | None = None)

Bases: object

SWH storage datastore proxy, encompassing DB and object storage

Instantiate a storage instance backed by a PostgreSQL database and an objstorage.

When db is passed as a connection string, then this module automatically manages a connection pool between min_pool_conns and max_pool_conns. When db is an explicit psycopg2 connection, then min_pool_conns and max_pool_conns are ignored and the connection is used directly.

Parameters:

• db – either a libpq connection string, or a psycopg2 connection

• objstorage – configuration for the backend ObjStorage; if unset, use a NoopObjStorage

• min_pool_conns – min number of connections in the psycopg2 pool

• max_pool_conns – max number of connections in the psycopg2 pool

• journal_writer – configuration for the JournalWriter

• query_options –

  configuration for the sql connections; keys of the dict are the method names decorated with db_transaction() or db_transaction_generator() (eg. content_find()), and values are dicts (config_name, config_value) used to configure the sql connection for the method_name. For example, using:

  {"content_get": {"statement_timeout": 5000}}
  

  will override the default statement timeout for the content_get() endpoint from 500ms to 5000ms.

  See swh.core.db.common for more details.

current_version: int = 193

get_db()

put_db(db)

db()

get_flavor() → str

property flavor: str

check_config(*, check_write: bool) → bool

content_add(content: List[Content]) → Dict[str, int]

content_update(contents: List[Dict[str, Any]], keys: List[str] = []) → None

content_add_metadata(content: List[Content]) → Dict[str, int]

content_get_data(content: HashDict | bytes) → bytes | None

content_get_partition(partition_id: int, nb_partitions: int, page_token: str | None = None, limit: int = 1000) → PagedResult[Content, str]

content_get(contents: List[bytes], algo: str = 'sha1') → List[Content | None]

content_missing(contents: List[HashDict], key_hash: str = 'sha1') → Iterable[bytes]

content_missing_per_sha1(contents: List[bytes]) → Iterable[bytes]

content_missing_per_sha1_git(contents: List[bytes]) → Iterable[bytes]

content_find(content: HashDict) → List[Content]

content_get_random() → bytes

skipped_content_add(content: List[SkippedContent]) → Dict[str, int]

skipped_content_find(content: HashDict) → List[SkippedContent]

skipped_content_missing(contents: List[Dict[str, Any]]) → Iterable[Dict[str, Any]]

directory_add(directories: List[Directory]) → Dict[str, int]

directory_missing(directories: List[bytes]) → Iterable[bytes]

directory_ls(directory: bytes, recursive: bool = False) → Iterable[Dict[str, Any]]

directory_entry_get_by_path(directory: bytes, paths: List[bytes]) → Dict[str, Any] | None

directory_get_random() → bytes

directory_get_entries(directory_id: bytes, page_token: bytes | None = None, limit: int = 1000) → PagedResult[DirectoryEntry, str] | None

directory_get_raw_manifest(directory_ids: List[bytes]) → Dict[bytes, bytes | None]

directory_get_id_partition(partition_id: int, nb_partitions: int, page_token: str | None = None, limit: int = 1000) → PagedResult[bytes, str]

revision_add(revisions: List[Revision]) → Dict[str, int]

revision_missing(revisions: List[bytes]) → Iterable[bytes]

revision_get_partition(partition_id: int, nb_partitions: int, page_token: str | None = None, limit: int = 1000) → PagedResult[Revision, str]

revision_get(revision_ids: List[bytes], ignore_displayname: bool = False) → List[Revision | None]

revision_log(revisions: List[bytes], ignore_displayname: bool = False, limit: int | None = None) → Iterable[Dict[str, Any] | None]

revision_shortlog(revisions: List[bytes], limit: int | None = None) → Iterable[Tuple[bytes, Tuple[bytes, ...]] | None]

revision_get_random() → bytes

extid_get_from_extid(id_type: str, ids: List[bytes], version: int | None = None) → List[ExtID]

extid_get_from_target(target_type: ObjectType, ids: List[bytes], extid_type: str | None = None, extid_version: int | None = None) → List[ExtID]

extid_add(ids: List[ExtID]) → Dict[str, int]

release_add(releases: List[Release]) → Dict[str, int]

release_missing(releases: List[bytes]) → Iterable[bytes]

release_get(releases: List[bytes], ignore_displayname: bool = False) → List[Release | None]

release_get_partition(partition_id: int, nb_partitions: int, page_token: str | None = None, limit: int = 1000) → PagedResult[Release, str]

release_get_random() → bytes

snapshot_add(snapshots: List[Snapshot]) → Dict[str, int]

snapshot_missing(snapshots: List[bytes]) → Iterable[bytes]

snapshot_get(snapshot_id: bytes) → Dict[str, Any] | None

snapshot_get_id_partition(partition_id: int, nb_partitions: int, page_token: str | None = None, limit: int = 1000) → PagedResult[bytes, str]

snapshot_count_branches(snapshot_id: bytes, branch_name_exclude_prefix: bytes | None = None) → Dict[str | None, int] | None

snapshot_get_branches(snapshot_id: bytes, branches_from: bytes = b'', branches_count: int = 1000, target_types: List[str] | None = None, branch_name_include_substring: bytes | None = None, branch_name_exclude_prefix: bytes | None = None) → PartialBranches | None

snapshot_get_random() → bytes

snapshot_branch_get_by_name(snapshot_id: bytes, branch_name: bytes, follow_alias_chain: bool = True, max_alias_chain_length: int = 100) → SnapshotBranchByNameResponse | None

origin_visit_add(visits: List[OriginVisit]) → Iterable[OriginVisit]

origin_visit_status_add(visit_statuses: List[OriginVisitStatus]) → Dict[str, int]

origin_visit_status_get_latest(origin_url: str, visit: int, allowed_statuses: List[str] | None = None, require_snapshot: bool = False) → OriginVisitStatus | None

origin_visit_get(origin: str, page_token: str | None = None, order: ListOrder = ListOrder.ASC, limit: int = 10) → PagedResult[OriginVisit, str]

origin_visit_get_with_statuses(origin: str, allowed_statuses: List[str] | None = None, require_snapshot: bool = False, page_token: str | None = None, order: ListOrder = ListOrder.ASC, limit: int = 10) → PagedResult[OriginVisitWithStatuses, str]

origin_visit_find_by_date(origin: str, visit_date: datetime, type: str | None = None) → OriginVisit | None

origin_visit_get_by(origin: str, visit: int) → OriginVisit | None

origin_visit_get_latest(origin: str, type: str | None = None, allowed_statuses: List[str] | None = None, require_snapshot: bool = False) → OriginVisit | None

origin_visit_status_get(origin: str, visit: int, page_token: str | None = None, order: ListOrder = ListOrder.ASC, limit: int = 10) → PagedResult[OriginVisitStatus, str]

origin_visit_status_get_random(type: str) → OriginVisitStatus | None

origin_get(origins: List[str]) → List[Origin | None]

origin_get_by_sha1(sha1s: List[bytes]) → List[Dict[str, Any] | None]

origin_get_range(origin_from=1, origin_count=100)

origin_list(page_token: str | None = None, limit: int = 100) → PagedResult[Origin, str]

origin_search(url_pattern: str, page_token: str | None = None, limit: int = 50, regexp: bool = False, with_visit: bool = False, visit_types: List[str] | None = None) → PagedResult[Origin, str]

origin_count(url_pattern: str, regexp: bool = False, with_visit: bool = False) → int

origin_snapshot_get_all(origin_url: str) → List[bytes]

origin_add(origins: List[Origin]) → Dict[str, int]

object_find_by_sha1_git(ids: List[bytes]) → Dict[bytes, List[Dict]]

stat_counters()

refresh_stat_counters()

raw_extrinsic_metadata_add(metadata: List[RawExtrinsicMetadata]) → Dict[str, int]

raw_extrinsic_metadata_get(target: ExtendedSWHID, authority: MetadataAuthority, after: datetime | None = None, page_token: bytes | None = None, limit: int = 1000) → PagedResult[RawExtrinsicMetadata, str]

raw_extrinsic_metadata_get_by_ids(ids: List[bytes]) → List[RawExtrinsicMetadata]

metadata_fetcher_add(fetchers: List[MetadataFetcher]) → Dict[str, int]

metadata_fetcher_get(name: str, version: str) → MetadataFetcher | None

raw_extrinsic_metadata_get_authorities(target: ExtendedSWHID) → List[MetadataAuthority]

metadata_authority_add(authorities: List[MetadataAuthority]) → Dict[str, int]

metadata_authority_get(type: MetadataAuthorityType, url: str) → MetadataAuthority | None

object_find_recent_references(target_swhid: ExtendedSWHID, limit: int) → List[ExtendedSWHID]

object_references_add(references: List[ObjectReference]) → Dict[str, int]

clear_buffers(object_types: Sequence[str] = ()) → None

Do nothing

flush(object_types: Sequence[str] = ()) → Dict[str, int]

object_delete(swhids: List[ExtendedSWHID]) → Dict[str, int]

Delete objects from the storage

All skipped content objects matching the given SWHID will be removed, including those who have the same SWHID due to hash collisions.

Origin objects are removed alongside their associated origin visit and origin visit status objects.

Parameters:

swhids – list of SWHID of the objects to remove

Returns:

number of objects removed. Details of each key:

content:delete

Number of content objects removed

content:delete:bytes

Sum of the removed contents’ data length

skipped_content:delete

Number of skipped content objects removed

directory:delete

Number of directory objects removed

revision:delete

Number of revision objects removed

release:delete

Number of release objects removed

snapshot:delete

Number of snapshot objects removed

origin:delete

Number of origin objects removed

origin_visit:delete

Number of origin visit objects removed

origin_visit_status:delete

Number of origin visit status objects removed

ori_metadata:delete

Number of raw extrinsic metadata objects targeting an origin that have been removed

snp_metadata:delete

Number of raw extrinsic metadata objects targeting a snapshot that have been removed

rev_metadata:delete

Number of raw extrinsic metadata objects targeting a revision that have been removed

rel_metadata:delete

Number of raw extrinsic metadata objects targeting a release that have been removed

dir_metadata:delete

Number ef raw extrinsic metadata objects targeting a directory that have been removed

cnt_metadata:delete

Number of raw extrinsic metadata objects targeting a content that have been removed

emd_metadata:delete

Number of raw extrinsic metadata objects targeting a raw extrinsic metadata object that have been removed

Return type:

dict

extid_delete_for_target(target_swhids: List[CoreSWHID]) → Dict[str, int]

Delete ExtID objects from the storage

Parameters:

target_swhids – list of SWHIDs targeted by the ExtID objects to remove

Returns:

extid:delete: Number of ExtID objects removed

Return type:

Summary dict with the following keys and associated values