fastapi otel example for testevents blog post

otel-ex-http.py

import asyncio
import random
import fastapi
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

# Initialize OTEL
provider = TracerProvider()
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

app = fastapi.FastAPI()

@app.get("/foo")
async def foo():
    MESSAGE = "hello world"
    with tracer.start_as_current_span("foo") as span:
        # Pick a random number between 0 and 1
        num = random.random()
        # Add a random attribute
        span.set_attribute("random_number", num)
        span.set_attribute("message", MESSAGE)
        return {"message": MESSAGE, "random_number": num}

@app.post("/bar")
async def bar(request: fastapi.Request):
    MESSAGE = "hello bar"
    with tracer.start_as_current_span("bar") as span:
        # Extract the random number from the request json
        data = await request.json()
        num = data["random_number"]
        span.set_attribute("random_number", num)
        span.set_attribute("message", MESSAGE)
        if num < 0.5:
            return {"message": MESSAGE}
        else:
            # Simulate an error
            raise ValueError("oh no")
        
@app.get("/baz")
async def baz():
    MESSAGE = "goodbye world"
    with tracer.start_as_current_span("baz") as span:
        # Sleep for a random amount of time
        sleep_time = random.random()
        span.set_attribute("sleep_time", sleep_time)
        span.set_attribute("message", MESSAGE)
        await asyncio.sleep(sleep_time)
        return {"message": MESSAGE}

FastAPIInstrumentor.instrument_app(app)