corvix.ingestion
================

.. py:module:: corvix.ingestion

.. autoapi-nested-parse::

   GitHub notifications ingestion client.



Attributes
----------

.. autoapisummary::

   corvix.ingestion.logger
   corvix.ingestion._ENRICHABLE_SUBJECT_TYPES
   corvix.ingestion._CHECK_SUITE_PATH_SEGMENTS
   corvix.ingestion._RELEASE_PATH_SEGMENTS


Classes
-------

.. autoapisummary::

   corvix.ingestion.WebUrlEnricher
   corvix.ingestion.GitHubNotificationsClient


Functions
---------

.. autoapisummary::

   corvix.ingestion._as_json_object
   corvix.ingestion._coerce_json_value
   corvix.ingestion.resolve_web_urls
   corvix.ingestion._http_error_detail
   corvix.ingestion._retry_delay_seconds


Module Contents
---------------

.. py:data:: logger

.. py:data:: _ENRICHABLE_SUBJECT_TYPES
   :type:  frozenset[str]

.. py:data:: _CHECK_SUITE_PATH_SEGMENTS
   :value: 5


.. py:data:: _RELEASE_PATH_SEGMENTS
   :value: 5


.. py:function:: _as_json_object(value: corvix.types.JsonValue) -> corvix.types.JsonObject | None

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


.. py:function:: _coerce_json_value(value: object) -> corvix.types.JsonValue

.. py:class:: WebUrlEnricher

   Bases: :py:obj:`Protocol`


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


   .. py:method:: enrich_web_url(notification: corvix.domain.Notification) -> str | None

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



.. py:function:: 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.


.. py:class:: GitHubNotificationsClient

   Client for GitHub notifications API.


   .. py:attribute:: token
      :type:  str


   .. py:attribute:: api_base_url
      :type:  str
      :value: 'https://api.github.com'



   .. py:attribute:: account_id
      :type:  str
      :value: 'primary'



   .. py:attribute:: account_label
      :type:  str
      :value: 'Primary'



   .. py:method:: fetch_notifications(polling: corvix.config.PollingConfig) -> list[corvix.domain.Notification]

      Fetch notifications with pagination.



   .. py:method:: _fetch_page(polling: corvix.config.PollingConfig, page: int) -> list[corvix.types.JsonObject]


   .. py:method:: mark_thread_read(thread_id: str) -> None

      Mark a notification thread as read.



   .. py:method:: dismiss_thread(thread_id: str) -> None

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



   .. py:method:: enrich_web_url(notification: corvix.domain.Notification) -> str | None

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



   .. py:method:: _resolve_check_suite(subject_url: str, repository: str) -> str | None


   .. py:method:: _resolve_release(subject_url: str) -> str | None


   .. py:method:: fetch_json_url(url: str, timeout_seconds: float = 30.0) -> corvix.types.JsonValue

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



   .. py:method:: _build_url(path: str, query: dict[str, str]) -> str


   .. py:method:: _headers() -> dict[str, str]


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


   .. py:method:: _request_no_content(url: str, method: str) -> None


   .. py:method:: _request_no_content_with_backoff(url: str, method: str, max_attempts: int = 4) -> None

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



   .. py:method:: _validate_api_host(url: str) -> None


.. py:function:: _http_error_detail(error: urllib.error.HTTPError) -> str

.. py:function:: _retry_delay_seconds(error: urllib.error.HTTPError, attempt: int) -> float