pyncette.prometheus module

class pyncette.prometheus.MeteredRepository(metric_set: pyncette.prometheus.OperationMetricSet, inner_repository: pyncette.repository.Repository)

Bases: pyncette.repository.Repository

A wrapper for repository that exposes metrics to Prometheus

commit_task(utc_now: datetime.datetime, task: pyncette.task.Task, lease: NewType.<locals>.new_type) → None

Commits the task, which signals a successful run.

extend_lease(utc_now: datetime.datetime, task: pyncette.task.Task, lease: NewType.<locals>.new_type) → Optional[NewType.<locals>.new_type]

Extends the lease on the task. Returns the new lease if lease was still valid.

poll_dynamic_task(utc_now: datetime.datetime, task: pyncette.task.Task, continuation_token: Optional[NewType.<locals>.new_type] = None) → pyncette.model.QueryResponse

Queries the dynamic tasks for execution

poll_task(utc_now: datetime.datetime, task: pyncette.task.Task, lease: Optional[NewType.<locals>.new_type] = None) → pyncette.model.PollResponse

Polls the task to determine whether it is ready for execution

register_task(utc_now: datetime.datetime, task: pyncette.task.Task) → None

Registers a dynamic task

unlock_task(utc_now: datetime.datetime, task: pyncette.task.Task, lease: NewType.<locals>.new_type) → None

Unlocks the task, making it eligible for retries in case execution failed.

unregister_task(utc_now: datetime.datetime, task: pyncette.task.Task) → None

Deregisters a dynamic task implementation

class pyncette.prometheus.OperationMetricSet(operation_name: str, labels: List[str])

Bases: object

Collection of Prometheus metrics representing a logical operation

measure(**labels) → AsyncIterator[None]

An async context manager that measures the execution of the wrapped code

pyncette.prometheus.prometheus_fixture(app_context: pyncette.pyncette.PyncetteContext) → AsyncIterator[None]

pyncette.prometheus.prometheus_middleware(context: pyncette.model.Context, next: Callable[[], Awaitable[None]]) → None

Middleware that exposes task execution metrics to Prometheus

pyncette.prometheus.use_prometheus(app: pyncette.pyncette.Pyncette, measure_repository: bool = True, measure_ticks: bool = True, measure_tasks: bool = True) → None

Decorate Pyncette app with Prometheus metric exporter.

Parameters:

• measure_repository – Whether to measure repository operations
• measure_ticks – Whether to measure ticks
• measure_tasks – Whether to measure individual task executions

pyncette.prometheus.with_prometheus_repository(repository_factory: pyncette.repository.RepositoryFactory) → pyncette.repository.RepositoryFactory

Wraps the repository factory into one that exposes the metrics via Prometheus