maple-python-style

Maple Python style

Acquire OTel objects at module scope.

from opentelemetry import metrics, trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("orders.api")
meter = metrics.get_meter("orders.api")

orders_submitted = meter.create_counter("orders.submitted", unit="1")

Bounded work

Prefer decorators for functions with clear boundaries.

@tracer.start_as_current_span("order.submit")
async def submit_order(*, tenant_id: str, order_id: str) -> None:
    span = trace.get_current_span()
    span.set_attributes({
        "tenant.id": tenant_id,
        "order.id": order_id,
    })

Use a context manager when a decorator does not fit.

with tracer.start_as_current_span("order.validate") as span:
    span.set_attribute("tenant.id", tenant_id)
    validate_order(order)

Do not use detached tracer.start_span(...); span.end() for bounded work.

Error paths

Record exceptions on the active span.

try:
    result = await client.messages.create(...)
except Exception as exc:
    span = trace.get_current_span()
    span.record_exception(exc)
    span.set_status(Status(StatusCode.ERROR))
    logger.exception("llm call failed", extra={"tenant_id": tenant_id})
    raise

Logs

If logs are claimed as OTLP-forwarded, configure all of:

• An OTel LoggerProvider + OTLPLogExporter + LoggingHandler
• set_logger_provider(logger_provider) from opentelemetry._logs
• Log correlation for existing records, e.g. LoggingInstrumentor().instrument(set_logging_format=True)

Preserve existing logging.basicConfig, console / file handlers, and log levels. The user's logger keeps working — you're adding an OTLP handler underneath so log lines carry trace_id / span_id and reach Maple.

Init behavior

Inline the endpoint and ingest key directly in the init module — don't read them from env. The ingest key is project-scoped + write-only (Sentry DSN shaped), so source-level configuration is the right default; env indirection just adds a class of "OTel didn't start because env wasn't set" deploy failures.

# telemetry.py
import logging

from opentelemetry import _logs, metrics, trace
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.logging import LoggingInstrumentor
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

MAPLE_ENDPOINT = "https://ingest.maple.dev"
MAPLE_KEY = "MAPLE_TEST"  # set by maple-onboard skill on pairing

_INITIALIZED = False


def init_observability() -> None:
    global _INITIALIZED
    if _INITIALIZED:
        return
    _INITIALIZED = True

    headers = {"authorization": f"Bearer {MAPLE_KEY}"}
    resource = Resource.create({
        "service.name": "my-python-app",
        "deployment.environment.name": os.getenv("DEPLOYMENT_ENV", "development"),
        "vcs.repository.url.full": "https://github.com/acme/my-python-app",
        "vcs.ref.head.revision": os.getenv("RAILWAY_GIT_COMMIT_SHA")
            or os.getenv("GITHUB_SHA")
            or os.getenv("GIT_COMMIT"),
    })

    tracer_provider = TracerProvider(resource=resource)
    tracer_provider.add_span_processor(
        BatchSpanProcessor(
            OTLPSpanExporter(endpoint=f"{MAPLE_ENDPOINT}/v1/traces", headers=headers),
        ),
    )
    trace.set_tracer_provider(tracer_provider)

    logger_provider = LoggerProvider(resource=resource)
    logger_provider.add_log_record_processor(
        BatchLogRecordProcessor(
            OTLPLogExporter(endpoint=f"{MAPLE_ENDPOINT}/v1/logs", headers=headers),
        ),
    )
    _logs.set_logger_provider(logger_provider)
    logging.getLogger().addHandler(LoggingHandler(logger_provider=logger_provider))
    LoggingInstrumentor().instrument(set_logging_format=True)

    meter_provider = MeterProvider(
        resource=resource,
        metric_readers=[
            PeriodicExportingMetricReader(
                OTLPMetricExporter(
                    endpoint=f"{MAPLE_ENDPOINT}/v1/metrics", headers=headers,
                ),
            ),
        ],
    )
    metrics.set_meter_provider(meter_provider)

Add the _INITIALIZED guard only when the app can realistically call this function more than once (FastAPI lifespan + workers, pytest fixtures, etc.).

Metrics

Counters:

• llm.tokens.input
• llm.tokens.output
• requests/events/jobs/errors

Use semantic units when the SDK supports them: token counters use unit="tokens". Do not add app-side llm.cost_usd pricing metrics for normal LLM calls; Maple estimates cost centrally from provider/model/token data.

Histograms:

• duration
• latency
• payload size

Avoid raw high-cardinality values in metric attributes. Prefer tenant/org/project, operation/use case, provider/model, and outcome dimensions over user-level metric tags.

FastAPI

Use the native instrumentation rather than replacing request handling with manual middleware.

from fastapi import FastAPI
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

import telemetry

telemetry.init_observability()
app = FastAPI()
FastAPIInstrumentor.instrument_app(app)

Import telemetry (and call init_observability()) before any other module that needs tracing.