Skip to content

API

pico_otel.config

Settings for pico-otel (prefix otel, zero-config).

OtelSettings dataclass

traces_exporter: auto (otlp if endpoint set, else console), otlp, console or none.

Source code in src/pico_otel/config.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@configured(target="self", prefix="otel", mapping="tree")
@dataclass
class OtelSettings:
    """``traces_exporter``: ``auto`` (otlp if ``endpoint`` set, else console),
    ``otlp``, ``console`` or ``none``."""

    enabled: bool = True
    service_name: str = "pico-app"
    endpoint: str = ""
    traces_exporter: str = "auto"
    traces_sample_ratio: float = 1.0
    prometheus_metrics: bool = True
    instrument_logging: bool = True
    instrument_sqlalchemy: bool = True
    instrument_celery: bool = True
    instrument_fastapi: bool = True

pico_otel.bootstrap

Wires the OpenTelemetry SDK and available instrumentors at container startup.

Everything optional is import-guarded: each instrumentor activates only when its package is installed (extras: fastapi, sqlalchemy, celery, otlp, prometheus). Setup is idempotent — a second container in the same process reuses the already-installed providers.

OtelBootstrap

Idempotent SDK setup + auto-instrumentation of installed libraries.

Source code in src/pico_otel/bootstrap.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@component
class OtelBootstrap:
    """Idempotent SDK setup + auto-instrumentation of installed libraries."""

    def __init__(self, settings: OtelSettings):
        self.settings = settings

    @configure
    def setup(self) -> None:
        s = self.settings
        if not s.enabled:
            return
        if not _already_configured():
            resource = Resource.create({"service.name": s.service_name})
            provider = TracerProvider(resource=resource, sampler=TraceIdRatioBased(s.traces_sample_ratio))
            exporter = _make_exporter(s)
            if exporter is not None:
                provider.add_span_processor(BatchSpanProcessor(exporter))
            trace.set_tracer_provider(provider)
            self._setup_metrics(resource)
        self._instrument()

    def _setup_metrics(self, resource) -> None:
        if not self.settings.prometheus_metrics:
            return
        try:
            from opentelemetry.exporter.prometheus import PrometheusMetricReader
            from opentelemetry.sdk.metrics import MeterProvider
        except ImportError:
            logger.info("prometheus metrics skipped: pip install pico-otel[prometheus]")
            return
        # PrometheusMetricReader writes to prometheus_client's default
        # registry — the same one pico-actuator serves at /actuator/metrics.
        metrics.set_meter_provider(MeterProvider(resource=resource, metric_readers=[PrometheusMetricReader()]))

    def _instrument(self) -> None:
        s = self.settings
        if s.instrument_logging:
            self._apply("opentelemetry.instrumentation.logging", "LoggingInstrumentor")
        if s.instrument_sqlalchemy:
            self._apply("opentelemetry.instrumentation.sqlalchemy", "SQLAlchemyInstrumentor")
        if s.instrument_celery:
            self._apply("opentelemetry.instrumentation.celery", "CeleryInstrumentor")

    @staticmethod
    def _apply(module: str, cls_name: str) -> None:
        try:
            mod = __import__(module, fromlist=[cls_name])
        except ImportError:
            return
        instrumentor = getattr(mod, cls_name)()
        if not instrumentor.is_instrumented_by_opentelemetry:
            instrumentor.instrument()

OtelFastApiConfigurer

Instruments the FastAPI app; collected structurally by pico-fastapi via List[FastApiConfigurer] (no import of pico-fastapi needed).

Source code in src/pico_otel/bootstrap.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@component
class OtelFastApiConfigurer:
    """Instruments the FastAPI app; collected structurally by pico-fastapi
    via ``List[FastApiConfigurer]`` (no import of pico-fastapi needed)."""

    priority = -100

    def __init__(self, settings: OtelSettings):
        self.settings = settings

    def configure_app(self, app) -> None:
        if not (self.settings.enabled and self.settings.instrument_fastapi):
            return
        try:
            from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
        except ImportError:
            logger.info("fastapi instrumentation skipped: pip install pico-otel[fastapi]")
            return
        FastAPIInstrumentor.instrument_app(app)