FastAPI with OpenTelemetry

app.py

from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from opentelemetry.context import get_current as get_current_context
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.trace import TracerProvider, _Span
from opentelemetry.sdk.trace.export import (BatchSpanProcessor,
                                            ConsoleSpanExporter,
                                            SimpleSpanProcessor)
from starlette.exceptions import HTTPException as StarletteHTTPException

from .models import Settings

app = FastAPI()

settings = Settings(_env_file=".env")
tracer = TracerProvider()

import .routes  # NOQA


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc):
    span_key = list(get_current_context().keys())[0]
    span = get_current_context().get(span_key)
    assert isinstance(span, _Span)
    return JSONResponse(
        {
            "message": str(exc.detail), 
            "trace_id": span.context.trace_id, 
            "span_id": span.context.span_id
        }, 
        status_code=exc.status_code
    )


@app.on_event("startup")
def startup_event():
    if settings.verbose_tracing:  # Prints all traces on console
        tracer.add_span_processor(
            SimpleSpanProcessor(ConsoleSpanExporter())
        )
    # Example of an external exporter, see https://opentelemetry.io/registry/?language=python
    if settings.app_insights_connection_string:
        exporter = AzureMonitorTraceExporter.from_connection_string(
            settings.app_insights_connection_string
        )
        tracer.add_span_processor(
            BatchSpanProcessor(exporter)
        )

    FastAPIInstrumentor.instrument_app(app, tracer_provider=tracer)