corvix.domain

Domain models for GitHub notifications.

Attributes

STORED_RECORD_LABEL

Classes

AccountError	Per-account fetch failure recorded during a poll cycle.
PollerStatus	Status written to the cache by the poller/watch loop.
Notification	Normalized notification from GitHub's notifications API.
NotificationRecord	Scored and rule-evaluated notification persisted for dashboards.

Functions

_as_object_map(→ dict[str, object] | None)
_require_non_empty_str(→ str)
_optional_str(→ str | None)
_optional_bool(→ bool)
_optional_float(→ float)
_optional_str_list(→ list[str])
_optional_context(→ dict[str, object])
_require_object_map(→ dict[str, object])
_require_non_empty_str_value(→ str)
parse_timestamp(→ datetime.datetime)	Parse an ISO8601 timestamp into a timezone-aware datetime.
format_timestamp(→ str)	Format a timezone-aware datetime as an ISO8601 string.
notification_key(→ str)	Return a stable account-scoped key for a notification.
_get_non_empty_str_or_default(→ str)

Module Contents

class corvix.domain.AccountError

Per-account fetch failure recorded during a poll cycle.

account_id: str

account_label: str

error: str

class corvix.domain.PollerStatus

Status written to the cache by the poller/watch loop.

status: str = 'unknown'

last_poll_time: str | None = None

last_error: str | None = None

last_error_time: str | None = None

account_errors: tuple[AccountError, Ellipsis] = ()

corvix.domain.STORED_RECORD_LABEL = 'stored record'

corvix.domain._as_object_map(value: object) → dict[str, object] | None

corvix.domain._require_non_empty_str(payload: collections.abc.Mapping[str, object], key: str, label: str) → str

corvix.domain._optional_str(payload: collections.abc.Mapping[str, object], key: str) → str | None

corvix.domain._optional_bool(payload: collections.abc.Mapping[str, object], key: str, default: bool, label: str) → bool

corvix.domain._optional_float(payload: collections.abc.Mapping[str, object], key: str, default: float, label: str) → float

corvix.domain._optional_str_list(payload: collections.abc.Mapping[str, object], key: str) → list[str]

corvix.domain._optional_context(payload: collections.abc.Mapping[str, object], key: str) → dict[str, object]

corvix.domain._require_object_map(value: object, message: str) → dict[str, object]

corvix.domain._require_non_empty_str_value(value: object, message: str) → str

corvix.domain.parse_timestamp(value: str) → datetime.datetime

Parse an ISO8601 timestamp into a timezone-aware datetime.

corvix.domain.format_timestamp(value: datetime.datetime) → str

Format a timezone-aware datetime as an ISO8601 string.

class corvix.domain.Notification

Normalized notification from GitHub’s notifications API.

thread_id: str

repository: str

reason: str

subject_title: str

subject_type: str

unread: bool

updated_at: datetime.datetime

thread_url: str | None = None

subject_url: str | None = None

web_url: str | None = None

repository_url: str | None = None

account_id: str = 'primary'

account_label: str = 'Primary'

classmethod from_api_payload(payload: collections.abc.Mapping[str, object], *, account_id: str = 'primary', account_label: str = 'Primary') → Notification

Build a notification from a GitHub API response payload.

class corvix.domain.NotificationRecord

Scored and rule-evaluated notification persisted for dashboards.

notification: Notification

score: float

excluded: bool

matched_rules: tuple[str, Ellipsis] = ()

actions_taken: tuple[str, Ellipsis] = ()

dismissed: bool = False

context: collections.abc.Mapping[str, object]

to_dict() → dict[str, object]

Convert to a JSON-serializable dictionary.

classmethod from_dict(payload: collections.abc.Mapping[str, object]) → NotificationRecord

Parse a stored record.

corvix.domain.notification_key(notification: Notification) → str

Return a stable account-scoped key for a notification.

corvix.domain._get_non_empty_str_or_default(payload: collections.abc.Mapping[str, object], key: str, default: str) → str