Emit a quality score to Prometheus¶

Surface a /metrics endpoint that exposes one counter for failures and one gauge for the latest score per contract. Requires pip install pycharter[server].

from http.server import BaseHTTPRequestHandler, HTTPServer

from pycharter import (
    QualityPrometheusObserver,
    QualityThresholds,
    SlackWebhookSink,
    check_quality,
    notify_on_breach,
)

observer = QualityPrometheusObserver()
slack = SlackWebhookSink(webhook_url="https://hooks.slack.invalid/T/B/abc")
thresholds = QualityThresholds(min_overall_score=95.0, max_violation_rate=0.05)


def evaluate(records: list[dict], *, contract_id: str) -> None:
    report = check_quality(
        data=records,
        contract_file=f"contracts/{contract_id}.yaml",
        thresholds=thresholds,
    )
    observer.record(report, contract=contract_id)
    notify_on_breach(report, contract=contract_id, sinks=[slack])


class MetricsHandler(BaseHTTPRequestHandler):
    def do_GET(self):  # noqa: N802 — http.server convention
        if self.path != "/metrics":
            self.send_response(404)
            self.end_headers()
            return
        body = observer.render_metrics()
        self.send_response(200)
        self.send_header("Content-Type", "text/plain; version=0.0.4")
        self.end_headers()
        self.wfile.write(body)


HTTPServer(("0.0.0.0", 9100), MetricsHandler).serve_forever()

Metrics exposed (all labelled by contract):

Name	Type	What it tracks
`pycharter_quality_records_total`	counter	Records evaluated
`pycharter_quality_violations_total`	counter	Records that failed validation
`pycharter_quality_threshold_breaches_total`	counter	`QualityThresholds` breaches
`pycharter_quality_score`	gauge	Latest overall score (0–100)
`pycharter_quality_completeness`	gauge	Latest completeness (0–1)
`pycharter_quality_accuracy`	gauge	Latest accuracy (0–1)
`pycharter_quality_violation_rate`	gauge	Latest violation rate (0–1)

To share a single /metrics endpoint with pycharter.streaming's observer, pass the same CollectorRegistry to both:

from prometheus_client import CollectorRegistry

registry = CollectorRegistry()
quality_observer = QualityPrometheusObserver(registry=registry)
# streaming_observer = PrometheusObserver(registry=registry)

Other available sinks: GenericWebhookSink (any URL), PagerDutyWebhookSink (Events API v2).