corvix.services

Application services orchestrating ingestion, actions, and rendering.

Classes

NotificationsClient	Client capabilities required by the poll cycle orchestration.
PollingSummary	Result of one polling cycle.
PollCycleInput	All inputs required by run_poll_cycle().

Functions

run_poll_cycle(→ PollingSummary)	Fetch notifications, score/evaluate, optionally execute actions, and persist cache.
run_watch_loop(→ list[PollingSummary])	Run polling loop suitable for local daemon usage.
render_cached_dashboards(...)	Load persisted records and render dashboards.
_select_dashboards(→ list[corvix.config.DashboardSpec])
_build_enrichment_providers(...)

Module Contents

class corvix.services.NotificationsClient

Bases: corvix.actions.MarkReadGateway, corvix.enrichment.base.JsonFetchClient, Protocol

Client capabilities required by the poll cycle orchestration.

fetch_notifications(polling: corvix.config.PollingConfig) → list[corvix.domain.Notification]

Fetch notifications with configured polling options.

class corvix.services.PollingSummary

Result of one polling cycle.

fetched: int

excluded: int

actions_taken: int

errors: list[str] = []

dispatch: corvix.notifications.models.DispatchResult | None = None

class corvix.services.PollCycleInput

All inputs required by run_poll_cycle().

Attributes:

config: Application configuration. clients: GitHub notifications clients. cache: Persistent notification cache. apply_actions: If False, actions are recorded as dry-run only. now: Override the current time (useful for testing). notification_targets: Optional notification dispatch targets.

config: corvix.config.AppConfig

cache: corvix.storage.NotificationCache

clients: tuple[NotificationsClient, Ellipsis] = ()

client: NotificationsClient | None = None

apply_actions: bool = False

now: datetime.datetime | None = None

notification_targets: list[corvix.notifications.targets.base.NotificationTarget] | None = None

corvix.services.run_poll_cycle(input: PollCycleInput) → PollingSummary

Fetch notifications, score/evaluate, optionally execute actions, and persist cache.

If input.notification_targets is provided (and input.config.notifications.enabled is True) the poll cycle will detect newly-arrived unread notifications and fan-out delivery to each target after saving the snapshot.

corvix.services.run_watch_loop(config: corvix.config.AppConfig, cache: corvix.storage.NotificationCache, apply_actions: bool, clients: tuple[NotificationsClient, Ellipsis] | None = None, client: NotificationsClient | None = None, iterations: int | None = None) → list[PollingSummary]

Run polling loop suitable for local daemon usage.

corvix.services.render_cached_dashboards(config: corvix.config.AppConfig, cache: corvix.storage.NotificationCache, console: rich.console.Console, dashboard_name: str | None = None) → list[corvix.presentation.DashboardRenderResult]

Load persisted records and render dashboards.

corvix.services._select_dashboards(config: corvix.config.AppConfig, dashboard_name: str | None) → list[corvix.config.DashboardSpec]

corvix.services._build_enrichment_providers(config: corvix.config.AppConfig) → list[corvix.enrichment.base.EnrichmentProvider]