app.services.sources_view

Read-only views of source registrations and search facets.

Same rules as app.services.search (a CatalogStore inside the caller’s unit of work, plain dataclasses out, no FastAPI, no source adapter).

Public API: SourceSummary, FacetProject, FacetSource, CatalogFacets, is_stale(), config_error_code(), list_sources(), get_facets().

class app.services.sources_view.CatalogFacets(*, sources)

Bases: object

Filter options for the search UI.

Parameters:

sources (tuple[FacetSource, ...]) – Sources with their projects and counts.

sources: tuple[FacetSource, ...]
class app.services.sources_view.FacetProject(*, project, count)

Bases: object

Record count of one project within a source.

Parameters:

  • project (str) – Project path.

  • count (int) – Number of decision rows (published and candidate) in it; errata are not counted.

project: str
count: int
class app.services.sources_view.FacetSource(*, source_id, count, projects)

Bases: object

One source’s projects, for the filter select.

Parameters:

  • source_id (str) – Registration id.

  • count (int) – Total rows across the source’s projects.

  • projects (tuple[FacetProject, ...]) – Projects of this source, with their counts.

source_id: str
count: int
projects: tuple[FacetProject, ...]
class app.services.sources_view.SourceSummary(*, id, type, enabled, config_error, config_error_code, intake, published_ref, last_attempt_at, last_success_at, last_status, last_error, stale)

Bases: object

One source registration’s configuration and freshness.

Parameters:

  • id (str) – Registration id.

  • type (str) – Source type.

  • enabled (bool) – Whether the registration may be polled / pushed to.

  • config_error (str | None) – Why the registration is disabled, if invalid.

  • config_error_code (ConfigErrorCode | None) – Stable code of config_error (see config_error_code()).

  • intake (tuple[IntakeKind, ...]) – Enabled intake mechanisms, from the current configuration (empty for a registration removed from catalog.yaml).

  • published_ref (str) – Configured published ref, from the current configuration (blank if removed from catalog.yaml).

  • 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) – Secret-free error details of the latest poll.

  • stale (bool) – Whether the indexed content may be out of date.

id: str
type: str
enabled: bool
config_error: str | None
config_error_code: ConfigErrorCode | None
intake: tuple[IntakeKind, ...]
published_ref: str
last_attempt_at: datetime | None
last_success_at: datetime | None
last_status: PollStatus
last_error: JsonValue
stale: bool
app.services.sources_view.config_error_code(registration, config)

Derive the stable code of a stored registration’s config error.

The stored config_error text is synced from the configuration loaded at startup (the same one passed here), so the code comes from that configuration; a stored registration missing from it was removed from catalog.yaml.

Parameters:
Return type:

ConfigErrorCode | None

Returns:

The code, or None without a config error.

app.services.sources_view.get_facets(store)

Build the source/project filter options with record counts.

Parameters:

store (CatalogStore) – Store of the caller’s unit of work.

Return type:

CatalogFacets

Returns:

The facets, sources ordered as list_facets() returns them.

app.services.sources_view.is_stale(registration, now, stale_after_minutes)

Tell whether a registration’s indexed content may be out of date.

Stale when the registration never succeeded, its latest poll failed or was partial, or its latest success is older than stale_after_minutes.

Parameters:

  • registration (Registration | None) – The registration, or None if unknown.

  • now (datetime) – The current time (injected for testability).

  • stale_after_minutes (int) – Configured staleness threshold.

Return type:

bool

Returns:

Whether the registration is stale.

app.services.sources_view.list_sources(store, config_sources, *, now, stale_after_minutes)

List every stored registration with its configuration and freshness.

Registrations removed from catalog.yaml are still listed (they are disabled, never deleted, per sync_registrations); their intake/published_ref are empty/blank since the current configuration no longer describes them.

Parameters:

  • store (CatalogStore) – Store of the caller’s unit of work.

  • config_sources (Sequence[SourceRegistration]) – The current configuration’s registrations.

  • now (datetime) – The current time (injected for testability).

  • stale_after_minutes (int) – Configured staleness threshold.

Return type:

tuple[SourceSummary, ...]

Returns:

One summary per stored registration, ordered by id.