JustAnalytics/Documentation

JustAnalytics for FastAPI

Step-by-step guide to add analytics, error tracking, and APM to your FastAPI app

JustAnalytics for FastAPI

Add distributed tracing, error tracking, structured logging, and infrastructure metrics to your FastAPI application.

Time to data: 3 minutes

Prerequisites#

Install the SDK

pip install justanalytics-python[fastapi]

Initialize and add middleware

# main.py
import os
import justanalytics
from fastapi import FastAPI
from justanalytics.integrations.fastapi import JustAnalyticsMiddleware

# Initialize the SDK
justanalytics.init(
    site_id=os.environ.get("JA_SITE_ID"),
    api_key=os.environ.get("JA_API_KEY"),
    service_name="fastapi-app",
    environment=os.environ.get("ENV", "development"),
)

app = FastAPI()

# Add middleware — traces all incoming requests automatically
app.add_middleware(JustAnalyticsMiddleware)

Add your endpoints

All endpoints are automatically traced. Each request creates a span named after the route pattern:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

class CreateUserRequest(BaseModel):
    name: str
    email: str

@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    # Automatically traced as "GET /api/users/{user_id}"
    user = await db.users.find_one(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.post("/api/users", status_code=201)
async def create_user(data: CreateUserRequest):
    user = await db.users.create(data.dict())
    return user

Add error handling

FastAPI HTTPException responses are tagged on the span. For capturing unexpected errors:

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    justanalytics.capture_exception(exc, tags={
        "path": request.url.path,
        "method": request.method,
    })
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
    )

Verify data is flowing

Start your server and make a test request:

uvicorn main:app --reload
curl http://localhost:8000/api/users/1

Verify Your Setup

After making a request, check the Traces tab. You should see spans with FastAPI route patterns like 'GET /api/users/{user_id}'.

Open Dashboard

Advanced Features#

Custom spans#

@app.get("/api/reports/{report_id}")
async def get_report(report_id: str):
    with justanalytics.start_span("generate-report") as span:
        span.set_attribute("report.id", report_id)

        with justanalytics.start_span("fetch-data"):
            data = await fetch_report_data(report_id)

        with justanalytics.start_span("render-pdf"):
            pdf = await render_pdf(data)

    return Response(content=pdf, media_type="application/pdf")

Dependency injection with tracing#

from fastapi import Depends

@justanalytics.span(op="db.query")
async def get_db_user(user_id: int):
    return await db.users.find_one(user_id)

@app.get("/api/users/{user_id}/orders")
async def get_user_orders(user: dict = Depends(get_db_user)):
    return await db.orders.find_many({"user_id": user["id"]})

User identification#

from fastapi import Depends, Security
from fastapi.security import HTTPBearer

security = HTTPBearer()

async def get_current_user(credentials = Security(security)):
    user = await verify_token(credentials.credentials)
    justanalytics.set_user(id=str(user.id), email=user.email)
    return user

Structured logging#

justanalytics.log("info", "Order placed", {"order_id": "o123", "total": 99.99})
justanalytics.log("error", "Payment failed", {"user_id": "u456", "reason": "card_expired"})

Background task tracing#

from fastapi import BackgroundTasks

async def send_welcome_email(user_id: str):
    with justanalytics.start_span("send-welcome-email") as span:
        span.set_attribute("user.id", user_id)
        await email_service.send(user_id, template="welcome")

@app.post("/api/users")
async def create_user(data: CreateUserRequest, background_tasks: BackgroundTasks):
    user = await db.users.create(data.dict())
    background_tasks.add_task(send_welcome_email, str(user.id))
    return user

Outgoing HTTP tracing#

from justanalytics.integrations.requests import RequestsIntegration

RequestsIntegration(service_name="fastapi-app").enable()

Lifecycle events for clean shutdown#

@app.on_event("shutdown")
async def shutdown_event():
    justanalytics.flush()
    justanalytics.close()

Custom metrics#

justanalytics.record_metric("custom.queue_depth", await get_queue_depth())
justanalytics.record_metric("custom.cache_hit_ratio", 0.87, {"cache": "redis"})

Environment variables#

| Variable | Description | Required | |----------|-------------|----------| | JA_SITE_ID | Your site ID from the dashboard | Yes | | JA_API_KEY | API key (starts with ja_sk_) | Yes | | ENV | Deployment environment | No |