mainframe.endpoints.opengrep
============================

.. py:module:: mainframe.endpoints.opengrep

.. autoapi-nested-parse::

   Staging-only OpenGrep shadow queue and result endpoints.



Attributes
----------

.. autoapisummary::

   mainframe.endpoints.opengrep.router
   mainframe.endpoints.opengrep.Authenticated
   mainframe.endpoints.opengrep.Database


Functions
---------

.. autoapisummary::

   mainframe.endpoints.opengrep.require_opengrep_shadow
   mainframe.endpoints.opengrep.get_opengrep_rules
   mainframe.endpoints.opengrep.dead_letter_expired_shadow_scans
   mainframe.endpoints.opengrep.queue_opengrep_alert
   mainframe.endpoints.opengrep.get_opengrep_jobs
   mainframe.endpoints.opengrep.submit_opengrep_result
   mainframe.endpoints.opengrep.get_unpublished_opengrep_results
   mainframe.endpoints.opengrep.require_publication_claim
   mainframe.endpoints.opengrep.checkpoint_opengrep_publication
   mainframe.endpoints.opengrep.heartbeat_opengrep_publication
   mainframe.endpoints.opengrep.acknowledge_opengrep_result


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

.. py:function:: require_opengrep_shadow() -> None

   Hide the shadow API unless staging explicitly enables it.


.. py:data:: router

.. py:data:: Authenticated

.. py:data:: Database

.. py:function:: get_opengrep_rules(state: Annotated[mainframe.rules.Rules, Depends(get_rules)]) -> mainframe.models.schemas.GetRules

   Return the reviewed OpenGrep corpus without exposing YARA rules.


.. py:function:: dead_letter_expired_shadow_scans(session: sqlalchemy.orm.Session, *, retry_before: datetime.datetime, now: datetime.datetime) -> None

   Fail expired shadow leases after the configured attempt budget.


.. py:function:: queue_opengrep_alert(package: mainframe.models.schemas.OpenGrepAlert, session: Database, auth: Authenticated) -> mainframe.models.schemas.QueuePackageResponse

   Create idempotent shadow work only for a package selected for alerting.


.. py:function:: get_opengrep_jobs(session: Database, auth: Authenticated, state: Annotated[mainframe.rules.Rules, Depends(get_rules)], batch: Annotated[int, Query(ge=1, le=100)] = 1) -> list[mainframe.models.schemas.JobResult]

   Lease OpenGrep work without consuming the canonical YARA queue.


.. py:function:: submit_opengrep_result(result: mainframe.models.schemas.OpenGrepScanResult | mainframe.models.schemas.OpenGrepScanResultFail, session: Database, auth: Authenticated) -> None

   Store a leased shadow result without changing the canonical scan.


.. py:function:: get_unpublished_opengrep_results(session: Database, limit: Annotated[int, Query(ge=1, le=100)] = 20) -> list[mainframe.models.schemas.OpenGrepResult]

   Atomically lease completed shadow results to one publisher.


.. py:function:: require_publication_claim(shadow: mainframe.models.orm.OpenGrepScan | None, publication_id: uuid.UUID) -> mainframe.models.orm.OpenGrepScan

   Return a matching terminal publication lease or raise a safe conflict.


.. py:function:: checkpoint_opengrep_publication(scan_id: uuid.UUID, progress: mainframe.models.schemas.OpenGrepPublicationProgress, session: Database) -> None

   Persist monotonic Discord thread progress for retry-safe resumption.


.. py:function:: heartbeat_opengrep_publication(scan_id: uuid.UUID, claim: mainframe.models.schemas.OpenGrepPublicationClaim, session: Database) -> None

   Renew an active publication lease without changing its progress.


.. py:function:: acknowledge_opengrep_result(scan_id: uuid.UUID, claim: mainframe.models.schemas.OpenGrepPublicationClaim, session: Database) -> mainframe.models.schemas.OpenGrepPublished

   Acknowledge publication only after the complete Discord thread exists.