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.

class swh.storage.postgresql.storage.Storage(db, objstorage, min_pool_conns=1, max_pool_conns=10, journal_writer=None, query_options=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

• 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 = 187

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: Union[HashDict, bytes]) → Optional[bytes]

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

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

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_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]) → Optional[Dict[str, Any]]

directory_get_random() → bytes

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

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

directory_get_id_partition(partition_id: int, nb_partitions: int, page_token: Optional[str] = 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: Optional[str] = None, limit: int = 1000) → PagedResult[Revision, str]

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

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

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

revision_get_random() → bytes

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

extid_get_from_target(target_type: ObjectType, ids: List[bytes], extid_type: Optional[str] = None, extid_version: Optional[int] = 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[Optional[Release]]

release_get_partition(partition_id: int, nb_partitions: int, page_token: Optional[str] = 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) → Optional[Dict[str, Any]]

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

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

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

snapshot_get_random() → bytes

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: Optional[List[str]] = None, require_snapshot: bool = False) → Optional[OriginVisitStatus]

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

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

origin_visit_find_by_date(origin: str, visit_date: datetime) → Optional[OriginVisit]

origin_visit_get_by(origin: str, visit: int) → Optional[OriginVisit]

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

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

origin_visit_status_get_random(type: str) → Optional[OriginVisitStatus]

origin_get(origins: List[str]) → Iterable[Optional[Origin]]

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

origin_get_range(origin_from=1, origin_count=100)

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

origin_search(url_pattern: str, page_token: Optional[str] = None, limit: int = 50, regexp: bool = False, with_visit: bool = False, visit_types: Optional[List[str]] = 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: Optional[datetime] = None, page_token: Optional[bytes] = 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) → Optional[MetadataFetcher]

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) → Optional[MetadataAuthority]

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

Do nothing

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