mainframe.server

Attributes

app

Functions

add_correlation(→ dict[str, Any])	Add request id to log message.
setup_logging(→ None)
monitor_queue(→ None)	Refresh queue metrics on a bounded interval independent of scrape traffic.
monitor_performance(→ None)	Refresh performance metrics independently of scrape traffic.
monitor_scan_cache(→ None)
monitor_ingestion(→ None)
lifespan(→ collections.abc.AsyncGenerator[None, None])	Load the state for the app.
metadata(→ mainframe.models.schemas.ServerMetadata)	Get server metadata.
update_rules(→ None)	Update the rules.

Module Contents

mainframe.server.add_correlation(_logger: logging.Logger, _method_name: str, event_dict: dict[str, Any]) → dict[str, Any]

Add request id to log message.

mainframe.server.setup_logging() → None

async mainframe.server.monitor_queue(monitor: mainframe.queue_monitor.QueueMonitor, refresh_seconds: int) → None

Refresh queue metrics on a bounded interval independent of scrape traffic.

async mainframe.server.monitor_performance(monitor: mainframe.performance_monitor.PerformanceMonitor, refresh_seconds: int) → None

Refresh performance metrics independently of scrape traffic.

async mainframe.server.monitor_scan_cache(app_: fastapi.FastAPI) → None

async mainframe.server.monitor_ingestion(pypi_client: mainframe.pypi.PyPIClient) → None

async mainframe.server.lifespan(app_: fastapi.FastAPI) → collections.abc.AsyncGenerator[None, None]

Load the state for the app.

mainframe.server.app

async mainframe.server.metadata() → mainframe.models.schemas.ServerMetadata

Get server metadata.

async mainframe.server.update_rules() → None

Update the rules.