`app.storage.port`¶

DB-neutral storage port of the catalog.

Services depend only on this module: the CatalogStoreProvider protocol hands out CatalogStore instances bound to one unit of work (one transaction). Everything crossing the port is a plain value type from app.storage.values (re-exported here) or app.domain; no ORM object leaks out.

Unit-of-work rules:

CatalogStoreProvider.unit_of_work() commits when the block exits normally and rolls back when it raises.
CatalogStoreProvider.lock_registration() is a unit of work that first takes an exclusive per-registration lock, held until the unit of work ends. All writes for one registration (poll, push) run inside it, which makes concurrent workers and replicas safe.
Conflicts are reported with domain errors from app.errors (DecisionExistsError, CandidateExistsError, DeliveryExistsError); they leave the unit of work usable.

Public API:

every value type of app.storage.values and app.storage.errata_values (re-exported here)
protocols CatalogStore, CatalogStoreProvider

class app.storage.port.CatalogStore(*args, **kwargs)¶

Bases: Protocol

Catalog operations inside one unit of work.

sync_registrations(registrations)¶

Make stored registrations match the configuration.

Listed registrations are inserted or updated; an invalid one gets status disabled, and a re-enabled one leaves disabled for never. Registrations missing from the list are disabled, never deleted.

Parameters:: registrations (Sequence[RegistrationConfig]) – All configured registrations.
Return type:: None

get_registration(source_id)¶

Fetch one registration.

Parameters:: source_id (str) – Registration id.
Return type:: Registration | None
Returns:: The registration, or None.

list_registrations()¶

List all registrations, enabled or not, ordered by id.

Return type:: list[Registration]
Returns:: The registrations.

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

Record the outcome of a poll.

last_success_at is set to attempted_at only for PollStatus.OK.

Parameters:

source_id (str) – Registration id.
status (PollStatus) – Poll outcome.
attempted_at (datetime) – When the poll started.
error (JsonValue) – Error details, None on success.

Return type:

None

put_revision(digest, content)¶

Store a snapshot unless it exists (snapshots are immutable).

Parameters:

digest (str) – Digest string.
content (str) – Canonical text content.

Return type:

bool

Returns:

True if it was newly stored.

get_revision(digest)¶

Fetch a snapshot.

Parameters:: digest (str) – Digest string.
Return type:: Revision | None
Returns:: The snapshot, or None.

upsert_published(record)¶

Insert or replace the published row at a location.

Keyed by (source_id, project, path).

Parameters:: record (RecordWrite) – Row content.
Return type:: UpsertResult
Returns:: The outcome.

upsert_candidate(record)¶

Insert or replace a candidate row.

Keyed by (source_id, project, record_key, ref).

Parameters:: record (RecordWrite) – Row content.
Return type:: UpsertResult
Returns:: The outcome.

create_candidate(record)¶

Insert a candidate row only if its key is free.

Parameters:: record (RecordWrite) – Row content.
Return type:: UpsertResult
Returns:: The outcome (always created).
Raises:: app.errors.CandidateExistsError – If the candidate (source_id, project, record_key, ref) exists.

get_record(record_id)¶

Fetch one record row.

Parameters:: record_id (UUID) – Record id.
Return type:: StoredRecord | None
Returns:: The row, or None.

list_change_tokens(source_id, project=None)¶

List the stored change state of published files.

Parameters:

source_id (str) – Registration id.
project (str | None) – Restrict to one project; None means all.

Return type:

list[ChangeTokenEntry]

Returns:

Entries ordered by project and path.

delete_published(record_ids)¶

Delete published rows (reconciliation). Candidates are kept.

Parameters:: record_ids (Collection[UUID]) – Ids to delete.
Return type:: int
Returns:: Number of deleted rows.

list_identity_siblings(identity)¶

List every row (published and candidates) of an identity.

Parameters:: identity (Identity) – Logical identity.
Return type:: list[StoredRecord]
Returns:: Rows, published first, then by ref and path.

list_relationship_summaries(kind=RecordState.PUBLISHED)¶

List relationship summaries, keyed by str(record id).

Parameters:: kind (RecordState | None) – Restrict to one state; None means all rows.
Return type:: list[RecordSummary]
Returns:: One summary per row.

list_relationship_candidates(self_uuid, link_targets)¶

List published rows that can affect one record’s relationships.

A bounded alternative to list_relationship_summaries( RecordState.PUBLISHED) for a single detail request: see app.storage.postgres.records.list_relationship_candidates() for exactly which rows this returns and why the result is equivalent for that one record’s view().

Parameters:

self_uuid (UUID) – The viewed record’s own UUID.
link_targets (Collection[UUID]) – UUIDs the viewed record’s own updates, obsoletes and corrects links target.

Return type:

list[RecordSummary]

Returns:

The matching published rows.

list_errata_for(target_uuid)¶

List the erratum rows whose corrects link is target_uuid.

Includes published rows and candidates (proposed errata), from every registration and project; resolving which decision an erratum applies to when the UUID is shared is up to the caller (app.domain.relationships rules). Decisions are never returned.

Parameters:: target_uuid (UUID) – UUID of the corrected decision.
Return type:: list[ErratumRow]
Returns:: Published rows first, then by source, project, path, ref and id.

list_published_by_uuid(uuids)¶

List the published rows holding any of uuids, with titles.

One query for a whole search page (erratum hits’ targets).

Parameters:: uuids (Collection[UUID]) – Authored UUIDs.
Return type:: list[LinkTarget]
Returns:: One entry per matching published row of any record kind (decisions and errata), with its full links.

list_facets()¶

Count decision rows per (source_id, project), for filter UIs.

Counts include both published and candidate rows, but not errata (matching the default record_type=decision search).

Return type:: list[FacetEntry]
Returns:: One entry per non-empty (source_id, project) pair, ordered by source id then project.

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

Full-text search with filters and bounded pagination.

Hits are ordered by sort (see SearchSort); the default ranks by relevance when query is given, else orders by title.

Parameters:

query (str | None) – Free text (titles, bodies, UUIDs, paths); blank or None matches everything.
filters (SearchFilters) – Filters.
page (int) – 1-based page number.
page_size (int) – Page size, 1 to MAX_PAGE_SIZE.
policy (Policy | None) – The configured policy (None when missing or invalid), used for approval states.
sort (SearchSort) – Result order; relevance without a query falls back to title.

Return type:

SearchPage

Returns:

The page.

Raises:

ValueError – If page or page_size is out of bounds.

append_decision(decision)¶

Append a decision event; the server sets id and timestamp.

Parameters:: decision (NewDecision) – The decision.
Return type:: DecisionEvent
Returns:: The stored event.
Raises:: app.errors.DecisionExistsError – If the identity already has a decision for the digest.

list_decisions(identity)¶

List all decision events of an identity, any digest.

Parameters:: identity (Identity) – Logical identity.
Return type:: list[DecisionEvent]
Returns:: Events, oldest first.

list_decisions_for(identities)¶

List all decision events of several identities in one query.

Parameters:: identities (Collection[Identity]) – Logical identities.
Return type:: list[DecisionEvent]
Returns:: Events of all of them, oldest first; group them by identity.

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

Record the start of an intake run.

Parameters:

source_id (str) – Registration id as requested.
kind (IntakeKind) – Poll or push.
identity (JsonValue) – Who triggered the run (no secrets).
delivery_id (str | None) – Push delivery id, if any.

Return type:

IntakeRun

Returns:

The running run.

Raises:

app.errors.DeliveryExistsError – If delivery_id was already used for source_id.

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

Record the end of an intake run.

Parameters:

run_id (UUID) – Run id.
status (IntakeStatus) – Final status.
counts (JsonValue) – Outcome counters.
errors (JsonValue) – Error details.

Return type:

IntakeRun

Returns:

The finished run.

Raises:

LookupError – If the run does not exist.

find_intake_run(source_id, delivery_id)¶

Find the run of an earlier push delivery.

Parameters:

source_id (str) – Registration id.
delivery_id (str) – Push delivery id.

Return type:

IntakeRun | None

Returns:

The run, or None.

class app.storage.port.CatalogStoreProvider(*args, **kwargs)¶

Bases: Protocol

Hands out CatalogStore instances per unit of work.

unit_of_work()¶

Open a unit of work (one transaction).

Return type:: AbstractContextManager[CatalogStore]
Returns:: A context manager yielding the store; it commits on normal exit and rolls back on error.

lock_registration(source_id)¶

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

The lock is taken before the store is yielded and released when the unit of work commits or rolls back. Other units of work for the same registration wait; other registrations are unaffected.

Parameters:: source_id (str) – Registration id.
Return type:: AbstractContextManager[CatalogStore]
Returns:: A context manager yielding the store.

class app.storage.port.ChangeTokenEntry(*, id, project, path, change_token, digest)¶

Bases: object

Stored change state of one published file.

Parameters:

id (UUID) – Record id.
project (str) – Owning project path.
path (str) – File path inside the project.
change_token (str | None) – Source change token, if known.
digest (str) – Current digest.

id: UUID¶

project: str¶

path: str¶

change_token: str | None¶

digest: str¶

class app.storage.port.DecisionEvent(*, identity, adr_uuid, digest, action, actor_issuer, actor_subject, actor_display, actor_email, comment, policy_id, policy_version, provenance=dict(), id, created_at)¶

Bases: NewDecision

A stored, append-only decision event.

Parameters:

id (UUID) – Stable event id.
created_at (datetime) – Server timestamp.

id: UUID¶

created_at: datetime¶

as_view()¶

Convert to the domain view used by approval evaluation.

Return type:: DecisionEventView
Returns:: The event view.

class app.storage.port.ErratumRow(*, id, kind, source_id, project, path, ref, adr_uuid, record_key, digest, title, erratum, updated_at, issues)¶

Bases: object

One erratum row (published or candidate) correcting a decision.

Parameters:

id (UUID) – Record id.
kind (RecordState) – Published or candidate.
source_id (str) – Registration id.
project (str) – Project path.
path (str) – File path inside the project.
ref (str) – Branch/ref of the content.
adr_uuid (UUID | None) – The erratum’s own UUID, if valid.
record_key (str) – Identity key of the erratum.
digest (str) – Current digest.
title (str) – Display title.
erratum (StoredErratum | None) – Stored content; None only for a row ingested before content was stored (the next ingest fills it).
updated_at (datetime) – When the content digest last changed.
issues (tuple[Issue, ...]) – Validation issues.

id: UUID¶

kind: RecordState¶

source_id: str¶

project: str¶

path: str¶

ref: str¶

adr_uuid: UUID | None¶

record_key: str¶

digest: str¶

title: str¶

erratum: StoredErratum | None¶

updated_at: datetime¶

issues: tuple[Issue, ...]¶

property identity: Identity¶

Logical identity of the erratum (approval is keyed by it).

Returns:: (source_id, project, record_key).

class app.storage.port.FacetEntry(*, source_id, project, count)¶

Bases: object

Row count for one (source_id, project) pair.

Used to populate the source/project filter select in the UI (app.storage.port.CatalogStore.list_facets()); counts include both published and candidate decision rows, never errata.

Parameters:

source_id (str) – Registration id.
project (str) – Project path.
count (int) – Number of rows for this source and project.

source_id: str¶

project: str¶

count: int¶

class app.storage.port.IntakeKind(*values)¶

Bases: StrEnum

How content entered the catalog.

POLL = 'poll'¶

PUSH = 'push'¶

class app.storage.port.IntakeRun(*, id, source_id, kind, identity, delivery_id, started_at, finished_at, status, counts, errors)¶

Bases: object

One poll or push intake run.

Parameters:

id (UUID) – Run id.
source_id (str) – Registration id as requested.
kind (IntakeKind) – Poll or push.
identity (JsonValue) – Who triggered the run (no secrets).
delivery_id (str | None) – Client-supplied push delivery id, if any.
started_at (datetime) – Start time.
finished_at (datetime | None) – End time, None while running.
status (IntakeStatus) – Run status.
counts (JsonValue) – Outcome counters.
errors (JsonValue) – Error details.

id: UUID¶

source_id: str¶

kind: IntakeKind¶

identity: JsonValue¶

delivery_id: str | None¶

started_at: datetime¶

finished_at: datetime | None¶

status: IntakeStatus¶

counts: JsonValue¶

errors: JsonValue¶

class app.storage.port.IntakeStatus(*values)¶

Bases: StrEnum

Status of one intake run.

RUNNING = 'running'¶

OK = 'ok'¶

PARTIAL = 'partial'¶

FAILED = 'failed'¶

type app.storage.port.JsonObject = Mapping[str, JsonValue]¶

class app.storage.port.LinkTarget(*, summary, title)¶

Bases: object

A published row that a link may resolve to, with its title.

Used to label erratum search hits (“Erratum for <title>”) without one query per hit.

Parameters:

summary (RecordSummary) – Relationship summary, keyed by str(record id).
title (str) – Display title of the row.

summary: RecordSummary¶

title: str¶

class app.storage.port.NewDecision(*, identity, adr_uuid, digest, action, actor_issuer, actor_subject, actor_display, actor_email, comment, policy_id, policy_version, provenance=dict())¶

Bases: object

A decision to append.

Parameters:

identity (Identity) – Logical identity decided on.
adr_uuid (UUID) – The record UUID.
digest (str) – Digest of the decided content.
action (DecisionAction) – Accepted or rejected.
actor_issuer (str) – Reviewer token issuer.
actor_subject (str) – Reviewer token subject.
actor_display (str | None) – Reviewer display name, if known.
actor_email (str | None) – Reviewer verified email, if known.
comment (str | None) – Optional reviewer comment.
policy_id (str) – Applied policy id.
policy_version (int) – Applied policy version.
provenance (JsonObject) – Record id, kind, ref, path and source URL at decision time.

identity: Identity¶

adr_uuid: UUID¶

digest: str¶

action: DecisionAction¶

actor_issuer: str¶

actor_subject: str¶

actor_display: str | None¶

actor_email: str | None¶

comment: str | None¶

policy_id: str¶

policy_version: int¶

provenance: JsonObject¶

class app.storage.port.PollStatus(*values)¶

Bases: StrEnum

Outcome of the latest poll of a registration.

NEVER = 'never'¶

OK = 'ok'¶

PARTIAL = 'partial'¶

FAILED = 'failed'¶

DISABLED = 'disabled'¶

class app.storage.port.RecordState(*values)¶

Bases: StrEnum

Lifecycle state of a stored record row.

PUBLISHED = 'published'¶

CANDIDATE = 'candidate'¶

class app.storage.port.RecordWrite(*, source_id, project, path, ref, adr_uuid, record_key, digest, change_token, metadata, title, title_source, decision_date, decision_date_inferred, record_kind, links, rendered_html, issues, source_url=None, author=None, erratum=None)¶

Bases: object

Everything ingest writes for one record row.

The revision for digest must have been stored first (app.storage.port.CatalogStore.put_revision()).

Parameters:

source_id (str) – Owning registration id.
project (str) – Owning project path.
path (str) – File path inside the project.
ref (str) – Branch/ref the content came from.
adr_uuid (UUID | None) – Authored UUID, None if missing or invalid.
record_key (str) – Identity key (app.domain.values.record_key()).
digest (str) – Digest string of the current content.
change_token (str | None) – Source change token (None for pushes).
metadata (JsonObject) – Parsed frontmatter.
title (str) – Display title.
title_source (TitleSource) – Where the title came from.
decision_date (date | None) – Displayed decision date.
decision_date_inferred (bool) – Whether the date is only inferred from the filename.
record_kind (RecordKind) – Decision or erratum.
links (Links) – Validated outbound links.
rendered_html (str) – Sanitized HTML rendering of the body.
issues (tuple[Issue, ...]) – Validation issues.
source_url (str | None) – Link to the content at the source, if any.
author (str | None) – Push or commit author, if known.
erratum (StoredErratum | None) – Rendered erratum content; None for decisions.

source_id: str¶

project: str¶

path: str¶

ref: str¶

adr_uuid: UUID | None¶

record_key: str¶

digest: str¶

change_token: str | None¶

metadata: JsonObject¶

title: str¶

title_source: TitleSource¶

decision_date: date | None¶

decision_date_inferred: bool¶

record_kind: RecordKind¶

links: Links¶

rendered_html: str¶

issues: tuple[Issue, ...]¶

source_url: str | None¶

author: str | None¶

erratum: StoredErratum | None¶

property identity: Identity¶

Logical identity of the record.

Returns:: (source_id, project, record_key).

class app.storage.port.Registration(*, id, type, config, enabled, config_error, last_attempt_at, last_success_at, last_status, last_error)¶

Bases: object

A stored source registration with its poll status.

Parameters:

id (str) – Stable registration id.
type (str) – Source type.
config (JsonObject) – Type-specific settings, secrets excluded.
enabled (bool) – Whether the registration may be polled / pushed to.
config_error (str | None) – Why the registration is disabled, if invalid.
last_attempt_at (datetime | None) – Start of the latest poll attempt.
last_success_at (datetime | None) – Start of the latest fully successful poll.
last_status (PollStatus) – Outcome of the latest poll.
last_error (JsonValue) – Error details of the latest poll, if any.

id: str¶

type: str¶

config: JsonObject¶

enabled: bool¶

config_error: str | None¶

last_attempt_at: datetime | None¶

last_success_at: datetime | None¶

last_status: PollStatus¶

last_error: JsonValue¶

class app.storage.port.RegistrationConfig(*, id, type, config, enabled=True, config_error=None)¶

Bases: object

One source registration as read from catalog.yaml.

Parameters:

id (str) – Stable registration id.
type (str) – Source type (filesystem or gitlab).
config (JsonObject) – Type-specific settings, secrets excluded.
enabled (bool) – False when the entry is invalid.
config_error (str | None) – Readable reason why the entry is invalid.

id: str¶

type: str¶

config: JsonObject¶

enabled: bool¶

config_error: str | None¶

class app.storage.port.Revision(*, digest, content, created_at)¶

Bases: object

An immutable, content-addressed snapshot.

Parameters:

digest (str) – Digest string (sha256-v1:...).
content (str) – Canonical text content.
created_at (datetime) – When the snapshot was first stored.

digest: str¶

content: str¶

created_at: datetime¶

class app.storage.port.SearchFilters(*, source_id=None, project=None, kind=None, needs_approval=False, record_kind=None)¶

Bases: object

Search filters; None / False means “no filter”.

Parameters:

source_id (str | None) – Registration id.
project (str | None) – Project path.
kind (RecordState | None) – Published or candidate.
needs_approval (bool) – Only records whose approval state needs approval (app.domain.approval.evaluate()).
record_kind (RecordKind | None) – Only decisions or only errata.

source_id: str | None¶

project: str | None¶

kind: RecordState | None¶

needs_approval: bool¶

record_kind: RecordKind | None¶

class app.storage.port.SearchItem(*, id, kind, source_id, project, path, ref, adr_uuid, record_key, digest, title, record_kind, decision_date, decision_date_inferred, issues, approval_state, corrects=None)¶

Bases: object

One search hit.

Parameters:

id (UUID) – Record id.
kind (RecordState) – Published or candidate.
source_id (str) – Registration id.
project (str) – Project path.
path (str) – File path inside the project.
ref (str) – Branch/ref of the content.
adr_uuid (UUID | None) – Authored UUID, if valid.
record_key (str) – Identity key.
digest (str) – Current digest.
title (str) – Display title.
record_kind (RecordKind) – Decision or erratum.
decision_date (date | None) – Displayed decision date.
decision_date_inferred (bool) – Whether the date is inferred.
issues (tuple[Issue, ...]) – Validation issues.
approval_state (ApprovalStateCode) – Approval state of the current digest, computed by the store with the same rules as app.domain.approval.evaluate().
corrects (UUID | None) – UUID an erratum corrects (None for decisions).

id: UUID¶

kind: RecordState¶

source_id: str¶

project: str¶

path: str¶

ref: str¶

adr_uuid: UUID | None¶

record_key: str¶

digest: str¶

title: str¶

record_kind: RecordKind¶

decision_date: date | None¶

decision_date_inferred: bool¶

issues: tuple[Issue, ...]¶

approval_state: ApprovalStateCode¶

corrects: UUID | None¶

class app.storage.port.SearchPage(*, items, total, page, page_size)¶

Bases: object

One page of search results.

Parameters:

items (tuple[SearchItem, ...]) – Hits on this page.
total (int) – Number of hits over all pages.
page (int) – 1-based page number.
page_size (int) – Requested page size.

items: tuple[SearchItem, ...]¶

total: int¶

page: int¶

page_size: int¶

class app.storage.port.SearchSort(*values)¶

Bases: StrEnum

Result order of app.storage.port.CatalogStore.search().

Every order ends with the lower-cased title and the record id, so pages are stable even when the leading key ties.

relevance: text rank, best first; without a text query it is the same as title.
title: lower-cased title A-Z.
recent: record.updated_at (when the content digest last changed), newest first.

RELEVANCE = 'relevance'¶

TITLE = 'title'¶

RECENT = 'recent'¶

class app.storage.port.StoredErratum(*, type, section, reported_by, reported_date, original_html, corrected_html, notes_html)¶

Bases: object

The stored content of one erratum record.

Parameters:

type (ErratumType | None) – Erratum type, None if missing or invalid.
section (str | None) – Where in the decision the error is, if given.
reported_by (str | None) – Free-text reporter; informational only, never an identity.
reported_date (date | None) – Authored report date, if valid.
original_html (str | None) – Sanitized HTML of “Original text”, if any.
corrected_html (str) – Sanitized HTML of “Corrected text”; empty when the section is missing.
notes_html (str | None) – Sanitized HTML of the notes, if any.

type: ErratumType | None¶

section: str | None¶

reported_by: str | None¶

reported_date: date | None¶

original_html: str | None¶

corrected_html: str¶

notes_html: str | None¶

class app.storage.port.StoredRecord(*, source_id, project, path, ref, adr_uuid, record_key, digest, change_token, metadata, title, title_source, decision_date, decision_date_inferred, record_kind, links, rendered_html, issues, source_url=None, author=None, erratum=None, id, kind, previous_digest, first_seen_at, updated_at, last_seen_at)¶