corvix.ingestion

GitHub notifications ingestion client.

Attributes

logger
_ENRICHABLE_SUBJECT_TYPES
_CHECK_SUITE_PATH_SEGMENTS
_RELEASE_PATH_SEGMENTS

Classes

WebUrlEnricher	Resolve web URLs for notifications where the fast path returned None.
GitHubNotificationsClient	Client for GitHub notifications API.

Functions

_as_json_object(→ corvix.types.JsonObject | None)	Return value as a JSON object when it is a dict.
_coerce_json_value(→ corvix.types.JsonValue)
resolve_web_urls(→ None)	Enrich web_url in-place for notifications where the fast path returned None.
_http_error_detail(→ str)
_retry_delay_seconds(→ float)

Module Contents

corvix.ingestion.logger

corvix.ingestion._ENRICHABLE_SUBJECT_TYPES: frozenset[str]

corvix.ingestion._CHECK_SUITE_PATH_SEGMENTS = 5

corvix.ingestion._RELEASE_PATH_SEGMENTS = 5

corvix.ingestion._as_json_object(value: corvix.types.JsonValue) → corvix.types.JsonObject | None

Return value as a JSON object when it is a dict.

corvix.ingestion._coerce_json_value(value: object) → corvix.types.JsonValue

class corvix.ingestion.WebUrlEnricher

Bases: Protocol

Resolve web URLs for notifications where the fast path returned None.

enrich_web_url(notification: corvix.domain.Notification) → str | None

Return a browser URL for the notification, or None if unresolvable.

corvix.ingestion.resolve_web_urls(notifications: list[corvix.domain.Notification], enricher: WebUrlEnricher | None = None) → None

Enrich web_url in-place for notifications where the fast path returned None.

class corvix.ingestion.GitHubNotificationsClient

Client for GitHub notifications API.

token: str

api_base_url: str = 'https://api.github.com'

account_id: str = 'primary'

account_label: str = 'Primary'

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

Fetch notifications with pagination.

_fetch_page(polling: corvix.config.PollingConfig, page: int) → list[corvix.types.JsonObject]

mark_thread_read(thread_id: str) → None

Mark a notification thread as read.

dismiss_thread(thread_id: str) → None

Dismiss a notification thread (removes it from inbox permanently).

enrich_web_url(notification: corvix.domain.Notification) → str | None

Resolve a browser URL via API for notification types the fast path cannot handle.

_resolve_check_suite(subject_url: str, repository: str) → str | None

_resolve_release(subject_url: str) → str | None

fetch_json_url(url: str, timeout_seconds: float = 30.0) → corvix.types.JsonValue

Fetch JSON from a fully-qualified API URL with host validation.

_build_url(path: str, query: dict[str, str]) → str

_headers() → dict[str, str]

_request_json(url: str, method: str, timeout_seconds: float = 30.0) → corvix.types.JsonValue

_request_no_content(url: str, method: str) → None

_request_no_content_with_backoff(url: str, method: str, max_attempts: int = 4) → None

Perform no-content request with retries for GitHub throttling responses.

_validate_api_host(url: str) → None

corvix.ingestion._http_error_detail(error: urllib.error.HTTPError) → str

corvix.ingestion._retry_delay_seconds(error: urllib.error.HTTPError, attempt: int) → float