corvix.ingestion

GitHub notifications ingestion client.

Attributes

logger
REQUEST_FAILED_DETAIL
_THREAD_ID_RE

Classes

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.
_validate_thread_id(→ None)	Raise ValueError if thread_id is not a valid GitHub notification thread ID.
_coerce_json_value(→ corvix.types.JsonValue)
_http_error_detail(→ str)
_retry_delay_seconds(→ float)

Module Contents

corvix.ingestion.logger

corvix.ingestion.REQUEST_FAILED_DETAIL = 'request failed'

corvix.ingestion._THREAD_ID_RE

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._validate_thread_id(thread_id: str) → None

Raise ValueError if thread_id is not a valid GitHub notification thread ID.

GitHub thread IDs are positive decimal integers. Rejecting anything that does not match prevents path-traversal sequences such as ../ from being embedded in the URL path constructed by callers.

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

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'

request_timeout_seconds: float = 30.0

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

Fetch notifications with pagination.

_fetch_page(polling: corvix.config.PollingConfig, page: int, timeout_seconds: float) → 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).

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]

_instrument_request(method: str) → collections.abc.Iterator[None]

Record GitHub API request count/latency metrics and a trace span.

Labels successful calls "success" and failed calls by HTTP status code (for HTTPError) or "error" (for other failures).

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

_request_no_content(url: str, method: str, timeout_seconds: float | None = None) → 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.

_sanitize_api_url(url: str) → str

Validate url and return a safe reconstruction using the trusted base host.

Reconstructs the URL with the scheme and netloc from self.api_base_url so that only the path and query from the input survive into the request. This neutralises SSRF via scheme injection or host-header manipulation while still allowing the caller to supply the full API path.

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

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