animedex.models.aggregate

Shared result envelope for multi-source aggregate commands.

Aggregate commands such as animedex season and animedex schedule fan out to multiple upstream backends and may receive a mix of successful rows and per-source failures. This module provides the stable envelope shape those commands return: items contains only rows from successful sources, while sources records one status row per selected backend.

AggregateSourceStatus

class animedex.models.aggregate.AggregateSourceStatus(*, backend: str, status: str, items: int = 0, reason: str | None = None, message: str | None = None, http_status: int | None = None, duration_ms: float = 0.0)[source]

Bases: AnimedexModel

Status row for one backend inside an aggregate response.

Variables:

  • backend (str) – Backend identifier, e.g. "anilist".

  • status (str) – "ok" for a successful source, "failed" for a source that raised while the fan-out continued.

  • items (int) – Number of successful rows contributed by this source.

  • reason (str or None) – Stable error reason when the source failed.

  • message (str or None) – Human-readable source failure message.

  • http_status (int or None) – HTTP status code when the failure exposed one.

  • duration_ms (float) – Wall-clock time spent in this source call.

backend: str
status: str
items: int
reason: str | None
message: str | None
http_status: int | None
duration_ms: float
property ok: bool

Return whether this source succeeded.

Returns:

True when status is "ok".

Return type:

bool

AggregateResult

class animedex.models.aggregate.AggregateResult(*, items: ~typing.List[~typing.Any] = <factory>, sources: ~typing.Dict[str, ~animedex.models.aggregate.AggregateSourceStatus] = <factory>, merge_diagnostics: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>)[source]

Bases: AnimedexModel

Envelope returned by multi-source aggregate commands.

The items list preserves each backend’s rich model. Failures are deliberately not injected into items; they live only in sources so a caller iterating over successful records never has to special-case failure sentinels.

Variables:

  • items (list) – Successful rows from every healthy source.

  • sources (dict[str, AggregateSourceStatus]) – Per-backend status map.

  • merge_diagnostics (list of dict) – Per-row diagnostics for rows that could not enter merge analysis.

items: List[Any]
sources: Dict[str, AggregateSourceStatus]
merge_diagnostics: List[Dict[str, Any]]
property failed_sources: Dict[str, AggregateSourceStatus]

Return the failed source statuses.

Returns:

Mapping containing only failed sources.

Return type:

dict[str, AggregateSourceStatus]

property succeeded_count: int

Return how many selected sources succeeded.

Returns:

Number of status == "ok" entries.

Return type:

int

property all_failed: bool

Return whether every selected source failed.

Returns:

True when at least one source was selected and none succeeded.

Return type:

bool

ScheduleCalendarResult

class animedex.models.aggregate.ScheduleCalendarResult(*, items: ~typing.List[~typing.Any] = <factory>, sources: ~typing.Dict[str, ~animedex.models.aggregate.AggregateSourceStatus] = <factory>, merge_diagnostics: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, timezone: str, window_start: ~datetime.date, window_end: ~datetime.date)[source]

Bases: AggregateResult

Aggregate schedule envelope with display-time metadata.

The JSON renderer emits this as a normal structured aggregate result. The TTY renderer uses the timezone and date window fields to group schedule rows into a calendar-like view.

Variables:

  • timezone (str) – IANA timezone name, "UTC", "local", or a fixed-offset value such as "+08:00".

  • window_start (datetime.date) – Inclusive local date for the schedule window.

  • window_end (datetime.date) – Exclusive local date for the schedule window.

timezone: str
window_start: date
window_end: date

MergedAnime

class animedex.models.aggregate.MergedAnime(*, title: ~animedex.models.anime.AnimeTitle, ids: ~typing.Dict[str, str] = <factory>, sources: ~typing.List[~animedex.models.common.SourceTag] = <factory>, records: ~typing.Dict[str, ~animedex.models.anime.Anime] = <factory>, core: ~typing.Dict[str, ~typing.Any] = <factory>, source_details: ~typing.Dict[str, ~typing.Dict[str, ~typing.Any]] = <factory>, source_payloads: ~typing.Dict[str, ~typing.Dict[str, ~typing.Any]] = <factory>, id_conflicts: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>)[source]

Bases: AnimedexModel

One anime entry merged across aggregate season sources.

Variables:

  • title (AnimeTitle) – Canonical display title chosen from the contributing records.

  • ids (dict[str, str]) – Combined external id map.

  • sources (list[SourceTag]) – Provenance tags for every contributing backend.

  • records (dict[str, Anime]) – Per-backend common anime projections.

  • core (dict) – Compact merged summary. The JSON output keeps this next to the full per-backend records so consumers can read the resolved item without recomputing it.

  • source_details (dict[str, dict]) – Per-backend source-specific fields promoted from the contributing records.

  • source_payloads (dict[str, dict]) – Full per-backend payloads for JSON consumers that need the complete upstream row shape.

  • id_conflicts (list of dict) – External id conflicts found while building the merged id map.

title: AnimeTitle
ids: Dict[str, str]
sources: List[SourceTag]
records: Dict[str, Anime]
core: Dict[str, Any]
source_details: Dict[str, Dict[str, Any]]
source_payloads: Dict[str, Dict[str, Any]]
id_conflicts: List[Dict[str, Any]]

selftest

animedex.models.aggregate.selftest() → bool[source]

Smoke-test the aggregate envelope model.

The diagnostic runner invokes this to confirm that nested rich models can be carried through the aggregate JSON path and that the source-status helpers behave correctly.

Returns:

True on success.

Return type:

bool