app.storage.postgres.store

PostgreSQL implementation of the storage port.

PostgresStoreProvider is constructed at the composition root with an engine; it opens one Session per unit of work and yields a PostgresCatalogStore bound to it. The store delegates each operation to a concern module (registrations, records, errata, search, events). All SQL uses bound parameters.

Registration locks use pg_advisory_xact_lock(namespace, hashtext(source_id)): the lock lives exactly as long as the transaction, so it is released by commit or rollback and works across processes and replicas.

Public API: LOCK_NAMESPACE, PostgresCatalogStore, PostgresStoreProvider.

app.storage.postgres.store.LOCK_NAMESPACE = 1094996481

First key of the two-key advisory lock; keeps registration locks apart from any other advisory lock users of the database.

class app.storage.postgres.store.PostgresCatalogStore(session)

Bases: object

CatalogStore bound to one session.

Instances are created by PostgresStoreProvider only; see CatalogStore for the operation contracts.

Parameters:

session (Session) – The unit-of-work session (inside a transaction).

acquire_registration_lock(source_id)

Take the registration’s transaction-scoped advisory lock.

Parameters:

source_id (str) – Registration id.

Return type:

None

sync_registrations(registrations)

See app.storage.port.CatalogStore.sync_registrations().

Return type:

None

get_registration(source_id)

See app.storage.port.CatalogStore.get_registration().

Return type:

Registration | None

list_registrations()

See app.storage.port.CatalogStore.list_registrations().

Return type:

list[Registration]

update_poll_status(source_id, status, attempted_at, error=None)

See app.storage.port.CatalogStore.update_poll_status().

Return type:

None

put_revision(digest, content)

See app.storage.port.CatalogStore.put_revision().

Return type:

bool

get_revision(digest)

See app.storage.port.CatalogStore.get_revision().

Return type:

Revision | None

upsert_published(record)

See app.storage.port.CatalogStore.upsert_published().

Return type:

UpsertResult

upsert_candidate(record)

See app.storage.port.CatalogStore.upsert_candidate().

Return type:

UpsertResult

create_candidate(record)

See app.storage.port.CatalogStore.create_candidate().

Return type:

UpsertResult

get_record(record_id)

See app.storage.port.CatalogStore.get_record().

Return type:

StoredRecord | None

list_change_tokens(source_id, project=None)

See app.storage.port.CatalogStore.list_change_tokens().

Return type:

list[ChangeTokenEntry]

delete_published(record_ids)

See app.storage.port.CatalogStore.delete_published().

Return type:

int

list_identity_siblings(identity)

See app.storage.port.CatalogStore.list_identity_siblings().

Return type:

list[StoredRecord]

list_relationship_summaries(kind=RecordState.PUBLISHED)

See CatalogStore.list_relationship_summaries().

Return type:

list[RecordSummary]

list_relationship_candidates(self_uuid, link_targets)

See CatalogStore.list_relationship_candidates().

Return type:

list[RecordSummary]

list_errata_for(target_uuid)

See app.storage.port.CatalogStore.list_errata_for().

Return type:

list[ErratumRow]

list_published_by_uuid(uuids)

See CatalogStore.list_published_by_uuid().

Return type:

list[LinkTarget]

list_facets()

See app.storage.port.CatalogStore.list_facets().

Return type:

list[FacetEntry]

search(query, filters, page, page_size, *, policy, sort=SearchSort.RELEVANCE)

See app.storage.port.CatalogStore.search().

Return type:

SearchPage

append_decision(decision)

See app.storage.port.CatalogStore.append_decision().

Return type:

DecisionEvent

list_decisions(identity)

See app.storage.port.CatalogStore.list_decisions().

Return type:

list[DecisionEvent]

list_decisions_for(identities)

See CatalogStore.list_decisions_for().

Return type:

list[DecisionEvent]

start_intake_run(source_id, kind, *, identity=None, delivery_id=None)

See app.storage.port.CatalogStore.start_intake_run().

Return type:

IntakeRun

finish_intake_run(run_id, status, *, counts=None, errors=None)

See app.storage.port.CatalogStore.finish_intake_run().

Return type:

IntakeRun

find_intake_run(source_id, delivery_id)

See app.storage.port.CatalogStore.find_intake_run().

Return type:

IntakeRun | None

class app.storage.postgres.store.PostgresStoreProvider(engine)

Bases: object

CatalogStoreProvider over an engine.

Parameters:

engine (Engine) – SQLAlchemy engine for the catalog database.

unit_of_work()

Open one session and transaction.

Return type:

Iterator[PostgresCatalogStore]

Returns:

A context manager yielding the store; commits on normal exit, rolls back on error.

lock_registration(source_id)

Open a unit of work holding the registration’s advisory lock.

Parameters:

source_id (str) – Registration id.

Return type:

Iterator[PostgresCatalogStore]

Returns:

A context manager yielding the store.