mainframe.endpoints.opengrep

Staging-only OpenGrep shadow queue and result endpoints.

Attributes

router
Authenticated
Database

Functions

require_opengrep_shadow(→ None)	Hide the shadow API unless staging explicitly enables it.
get_opengrep_rules(]) → mainframe.models.schemas.GetRules)	Return the reviewed OpenGrep corpus without exposing YARA rules.
dead_letter_expired_shadow_scans(→ None)	Fail expired shadow leases after the configured attempt budget.
queue_opengrep_alert(...)	Create idempotent shadow work only for a package selected for alerting.
get_opengrep_jobs(], batch, ...)	Lease OpenGrep work without consuming the canonical YARA queue.
submit_opengrep_result(→ None)	Store a leased shadow result without changing the canonical scan.
get_unpublished_opengrep_results(...)	Atomically lease completed shadow results to one publisher.
require_publication_claim(...)	Return a matching terminal publication lease or raise a safe conflict.
checkpoint_opengrep_publication(→ None)	Persist monotonic Discord thread progress for retry-safe resumption.
heartbeat_opengrep_publication(→ None)	Renew an active publication lease without changing its progress.
acknowledge_opengrep_result(...)	Acknowledge publication only after the complete Discord thread exists.

Module Contents

mainframe.endpoints.opengrep.require_opengrep_shadow() → None

Hide the shadow API unless staging explicitly enables it.

mainframe.endpoints.opengrep.router

mainframe.endpoints.opengrep.Authenticated

mainframe.endpoints.opengrep.Database

mainframe.endpoints.opengrep.get_opengrep_rules(state: Annotated[mainframe.rules.Rules, Depends(get_rules)]) → mainframe.models.schemas.GetRules

Return the reviewed OpenGrep corpus without exposing YARA rules.

mainframe.endpoints.opengrep.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.

mainframe.endpoints.opengrep.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.

mainframe.endpoints.opengrep.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.

mainframe.endpoints.opengrep.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.

mainframe.endpoints.opengrep.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.

mainframe.endpoints.opengrep.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.

mainframe.endpoints.opengrep.checkpoint_opengrep_publication(scan_id: uuid.UUID, progress: mainframe.models.schemas.OpenGrepPublicationProgress, session: Database) → None

Persist monotonic Discord thread progress for retry-safe resumption.

mainframe.endpoints.opengrep.heartbeat_opengrep_publication(scan_id: uuid.UUID, claim: mainframe.models.schemas.OpenGrepPublicationClaim, session: Database) → None

Renew an active publication lease without changing its progress.

mainframe.endpoints.opengrep.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.