EXAMPLE FastAPI

with new_context():
    identify_context(user.email)
    tag('email', user.email)
    tag('is_staff', user.is_staff)
    capture('user_signed_up', properties={'signup_method': 'form'})

with new_context():
    identify_context(user.email)
    tag('email', user.email)
    tag('is_staff', user.is_staff)
    capture('user_logged_in', properties={'login_method': 'password'})

with new_context():
    identify_context(current_user.email)
    capture('burrito_considered', properties={'total_considerations': count})

show_new_feature = posthog.feature_enabled(
    'new-dashboard-feature',
    current_user.email,
    person_properties={'email': current_user.email, 'is_staff': current_user.is_staff}
)
feature_config = posthog.get_feature_flag_payload('new-dashboard-feature', current_user.email)

try:
    # Critical operation that might fail
    result = process_payment()
except Exception as e:
    # Manually capture this specific exception
    with new_context():
        identify_context(current_user.email)
        event_id = posthog.capture_exception(e)

    return JSONResponse({
        "error": "Operation failed",
        "error_id": event_id,
        "message": f"Error captured in PostHog. Reference ID: {event_id}"
    }, status_code=500)

basics/fastapi/
├── app/
│   ├── __init__.py              # Package marker
│   ├── config.py                # Pydantic Settings configuration
│   ├── database.py              # SQLAlchemy setup
│   ├── dependencies.py          # FastAPI dependency injection
│   ├── main.py                  # Application factory and lifespan
│   ├── models.py                # User model (SQLAlchemy)
│   ├── routers/
│   │   ├── __init__.py          # Routers package
│   │   ├── main.py              # Page routes (HTML)
│   │   └── api.py               # API endpoints (JSON)
│   └── templates/               # Jinja2 templates
├── .env.example
├── .gitignore
├── requirements.txt
├── README.md
└── run.py                       # Entry point (uvicorn)

POSTHOG_PROJECT_TOKEN=<ph_project_token>
POSTHOG_HOST=https://us.i.posthog.com
SECRET_KEY=your-secret-key-here
DEBUG=True
POSTHOG_DISABLED=False

"""FastAPI PostHog example application."""

"""FastAPI application configuration using Pydantic Settings."""

from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    """Application settings loaded from environment variables."""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    # Application
    secret_key: str = "dev-secret-key-change-in-production"
    debug: bool = True

    # Database (SQLite like Flask example)
    database_url: str = "sqlite:///./db.sqlite3"

    # PostHog
    posthog_project_token: str = "<ph_project_token>"
    posthog_host: str = "https://us.i.posthog.com"
    posthog_disabled: bool = False


@lru_cache
def get_settings() -> Settings:
    """Get cached settings instance."""
    return Settings()

"""Database configuration with SQLAlchemy."""

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from app.config import get_settings

settings = get_settings()

engine = create_engine(
    settings.database_url,
    connect_args={"check_same_thread": False},  # Required for SQLite
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


class Base(DeclarativeBase):
    """Base class for SQLAlchemy models."""

    pass


def get_db():
    """Dependency that provides a database session."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


def init_db():
    """Create all database tables."""
    Base.metadata.create_all(bind=engine)

"""Authentication dependencies for FastAPI."""

from typing import Annotated, Optional

from fastapi import Cookie, Depends, HTTPException, status
from itsdangerous import BadSignature, URLSafeSerializer
from sqlalchemy.orm import Session

from app.config import get_settings
from app.database import get_db
from app.models import User

settings = get_settings()
serializer = URLSafeSerializer(settings.secret_key)


def get_session_user_id(session_token: Annotated[Optional[str], Cookie()] = None) -> Optional[int]:
    """Extract user ID from session cookie."""
    if not session_token:
        return None
    try:
        data = serializer.loads(session_token)
        return data.get("user_id")
    except BadSignature:
        return None


def get_current_user(
    db: Annotated[Session, Depends(get_db)],
    user_id: Annotated[Optional[int], Depends(get_session_user_id)],
) -> Optional[User]:
    """Get the current authenticated user, or None if not authenticated."""
    if user_id is None:
        return None
    return User.get_by_id(db, user_id)


def require_auth(
    current_user: Annotated[Optional[User], Depends(get_current_user)],
) -> User:
    """Require authentication - raises 401 if not authenticated."""
    if current_user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication required",
        )
    return current_user


def create_session_token(user_id: int) -> str:
    """Create a signed session token for the user."""
    return serializer.dumps({"user_id": user_id})


# Type aliases for cleaner dependency injection
CurrentUser = Annotated[Optional[User], Depends(get_current_user)]
RequiredUser = Annotated[User, Depends(require_auth)]
DbSession = Annotated[Session, Depends(get_db)]

"""FastAPI application with PostHog integration."""

from contextlib import asynccontextmanager
from pathlib import Path

import posthog
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from app.config import get_settings
from app.database import SessionLocal, init_db
from app.middleware import PostHogMiddleware
from app.models import User
from app.routers import api, main

settings = get_settings()

# Setup templates
templates_dir = Path(__file__).parent / "templates"
templates = Jinja2Templates(directory=str(templates_dir))


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan events for startup/shutdown."""
    # Startup: Initialize PostHog
    if not settings.posthog_disabled:
        posthog.api_key = settings.posthog_project_token
        posthog.host = settings.posthog_host
        posthog.debug = settings.debug

    # Initialize database and seed default user
    init_db()
    db = SessionLocal()
    try:
        if not User.get_by_email(db, "admin@example.com"):
            User.create_user(
                db,
                email="admin@example.com",
                password="admin",
                is_staff=True,
            )
    finally:
        db.close()

    yield

    # Shutdown: Flush PostHog events
    if not settings.posthog_disabled:
        posthog.flush()


app = FastAPI(
    title="PostHog FastAPI Example",
    description="Example application demonstrating PostHog integration with FastAPI",
    lifespan=lifespan,
)

app.add_middleware(PostHogMiddleware)

# Include routers
app.include_router(main.router)
app.include_router(api.router, prefix="/api")


# Error handlers
@app.exception_handler(404)
async def not_found_handler(request: Request, exc):
    """Handle 404 errors."""
    if request.url.path.startswith("/api/"):
        return JSONResponse({"error": "Not found"}, status_code=404)
    return templates.TemplateResponse(
        request, "errors/404.html", status_code=404
    )


@app.exception_handler(500)
async def internal_error_handler(request: Request, exc):
    """Handle 500 errors."""
    if request.url.path.startswith("/api/"):
        return JSONResponse({"error": "Internal server error"}, status_code=500)
    return templates.TemplateResponse(
        request, "errors/500.html", status_code=500
    )

"""PostHog middleware for automatic context and user identification.

Uses pure ASGI middleware instead of BaseHTTPMiddleware for better performance/best practices.
"""

from http.cookies import SimpleCookie
from typing import Callable, Optional

from posthog import identify_context, new_context, tag

from app.config import get_settings
from app.database import SessionLocal
from app.dependencies import serializer
from app.models import User


class PostHogMiddleware:
    """Pure ASGI middleware that wraps each request in a PostHog context.

    If the user is authenticated, identifies them in the context so routes
    can just call capture() without needing to set up context each time.

    Uses pure ASGI interface for better performance than BaseHTTPMiddleware.
    """

    def __init__(self, app):
        self.app = app
        self.settings = get_settings()

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http" or self.settings.posthog_disabled:
            await self.app(scope, receive, send)
            return

        user = self._get_user_from_scope(scope)

        with new_context():
            if user:
                identify_context(str(user.id))
                tag("is_staff", user.is_staff)

            await self.app(scope, receive, send)

    def _get_user_from_scope(self, scope) -> Optional[User]:
        """Extract authenticated user from session cookie in ASGI scope."""
        headers = dict(scope.get("headers", []))
        cookie_header = headers.get(b"cookie", b"").decode("utf-8")

        if not cookie_header:
            return None

        cookies = SimpleCookie()
        cookies.load(cookie_header)

        session_cookie = cookies.get("session_token")
        if not session_cookie:
            return None

        session_token = session_cookie.value

        try:
            data = serializer.loads(session_token)
            user_id = data.get("user_id")
        except Exception:
            return None

        if not user_id:
            return None

        db = SessionLocal()
        try:
            return User.get_by_id(db, user_id)
        finally:
            db.close()

"""User model with SQLite persistence (similar to Flask example)."""

from datetime import datetime, timezone
from typing import Optional

from sqlalchemy import Boolean, DateTime, Integer, String
from sqlalchemy.orm import Mapped, Session, mapped_column
from werkzeug.security import check_password_hash, generate_password_hash

from app.database import Base


class User(Base):
    """User model with SQLite persistence."""

    __tablename__ =

"""FastAPI routers package."""

"""API endpoints demonstrating PostHog integration patterns."""

from typing import Annotated

import posthog
from fastapi import APIRouter, Cookie, Form, Query
from fastapi.responses import JSONResponse
from posthog import capture

from app.dependencies import RequiredUser

router = APIRouter()

MAX_BURRITO_COUNT = 10000


@router.post("/burrito/consider")

"""Main routes demonstrating PostHog integration patterns."""

from pathlib import Path
from typing import Annotated

import posthog
from fastapi import APIRouter, Cookie, Depends, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from posthog import capture, identify_context, new_context

from app.dependencies import (
    CurrentUser,
    DbSession,

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>{% block title %}PostHog FastAPI Example{% endblock %}</title>
    <style>
        * {
            box-sizing

{% extends "base.html" %}

{% block title %}Burrito - PostHog FastAPI Example{% endblock %}

{% block content %}
<div class="card">
    <h1>Burrito Consideration Tracker</h1>
    <p>This page demonstrates custom event tracking with PostHog.</p>

    <div class="count" id="burrito-count">{{ burrito_count }}</div>
    <p style="text-align: center; color: #666;">Times you've considered a burrito</p>

    <div style="text-align: center; margin-top: 20px;">
        <button onclick="considerBurrito()">Consider a Burrito</button>
    </div>
</div>

<div class="card">
    <h3>Code Example</h3>
    <pre>
# API endpoint captures the event
with new_context():
    identify_context(current_user.email)
    capture('burrito_considered', properties={
        'total_considerations': burrito_count
    })</pre>
</div>
{% endblock %}

{% block scripts %}
<script>
async function considerBurrito() {
    try {
        const response = await fetch('/api/burrito/consider', {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json'
            }
        });
        const data = await response.json();
        if (data.success) {
            document.getElementById('burrito-count').textContent = data.count;
        }
    } catch (error) {
        console.error('Error:', error);
    }
}
</script>
{% endblock %}

{% extends "base.html" %}

{% block title %}Dashboard - PostHog FastAPI Example{% endblock %}

{% block content %}
<div class="card">
    <h1>Dashboard</h1>
    <p>Welcome back, {{ current_user.email }}!</p>
</div>

<div class="card">
    <h2>Feature Flags</h2>

    {% if show_new_feature %}
    <div class="feature-flag">
        <strong>New Feature Enabled!</strong>
        <p>You're seeing this because the <code>new-dashboard-feature</code> flag is enabled for you.</p>
        {% if feature_config %}
        <p><strong>Feature Configuration:</strong></p>
        <pre>{{ feature_config | tojson(indent=2) }}</pre>
        {% endif %}
    </div>
    {% else %}
    <p>The <code>new-dashboard-feature</code> flag is not enabled for your account.</p>
    {% endif %}

    <h3 style="margin-top: 20px;">Code Example</h3>
    <pre>
# Check if feature flag is enabled
show_new_feature = posthog.feature_enabled(
    'new-dashboard-feature',
    user_id,
    person_properties={
        'email': current_user.email,
        'is_staff': current_user.is_staff
    }
)

# Get feature flag payload
feature_config = posthog.get_feature_flag_payload(
    'new-dashboard-feature',
    user_id
)</pre>
</div>
{% endblock %}

{% extends "base.html" %}

{% block title %}Page Not Found - PostHog FastAPI Example{% endblock %}

{% block content %}
<div class="card">
    <h1>404 - Page Not Found</h1>
    <p>The page you're looking for doesn't exist.</p>
    <a href="/" class="btn">Go Home</a>
</div>
{% endblock %}

{% extends "base.html" %}

{% block title %}Server Error - PostHog FastAPI Example{% endblock %}

{% block content %}
<div class="card">
    <h1>500 - Internal Server Error</h1>
    <p>Something went wrong on our end. Please try again later.</p>
    <a href="/" class="btn">Go Home</a>
</div>
{% endblock %}

{% extends "base.html" %}

{% block title %}Login - PostHog FastAPI Example{% endblock %}

{% block content %}
<div class="card">
    <h1>Welcome to PostHog FastAPI Example</h1>
    <p>This example demonstrates how to integrate PostHog with a FastAPI application.</p>

    <form method="POST">
        <label for="email">Email</label>
        <input type="email" id="email" name="email" required>

        <label for="password">Password</label>
        <input type="password" id="password" name="password" required>

        <button type="submit">Login</button>
    </form>

    <p style="margin-top: 16px; font-size: 14px; color: #666;">
        Don't have an account? <a href="/signup">Sign up here</a>
    </p>
    <p style="font-size: 14px; color: #666;">
        <strong>Tip:</strong> Default credentials are admin@example.com/admin
    </p>
</div>

<div class="card">
    <h2>Features Demonstrated</h2>
    <ul style="margin-left: 20px; color: #666;">
        <li>User registration and identification</li>
        <li>Event tracking</li>
        <li>Feature flags</li>
        <li>Error tracking</li>
        <li>Group analytics</li>
    </ul>
</div>
{% endblock %}

{% extends "base.html" %}

{% block title %}Profile - PostHog FastAPI Example{% endblock %}

{% block content %}
<div class="card">
    <h1>Your Profile</h1>
    <p>This page demonstrates profile updates and report generation with PostHog.</p>

    {% if success %}
    <div class="message success">{{ success }}</div>
    {% endif %}

    <form method="POST" action

{% extends "base.html" %}

{% block title %}Sign Up - PostHog FastAPI Example{% endblock %}

{% block content %}
<div class="card">
    <h1>Create an Account</h1>
    <p>Sign up to explore the PostHog FastAPI integration example.</p>

    <form method="POST">
        <label for="email">Email *</label>
        <input type="email" id="email" name="email" required>

        <label for="password">Password *</label>
        <input type="password" id="password" name="password" required>

        <label for="password_confirm">Confirm Password *</label>
        <input type="password" id="password_confirm" name="password_confirm" required>

        <button type="submit">Sign Up</button>
    </form>

    <p style="margin-top: 16px; font-size: 14px; color: #666;">
        Already have an account? <a href="/">Login here</a>
    </p>
</div>

<div class="card">
    <h2>PostHog Integration</h2>
    <p>When you sign up, the following PostHog events are captured:</p>
    <ul style="margin-left: 20px; color: #666;">
        <li><code>identify_context()</code> - Associates your email with the context</li>
        <li><code>tag()</code> - Sets person properties (email, etc.)</li>
        <li><code>user_signed_up</code> event - Tracks the signup action</li>
    </ul>

    <h3 style="margin-top: 20px;">Code Example</h3>
    <pre>
# After creating the user
with new_context():
    identify_context(user.email)

    tag('email', user.email)
    tag('is_staff', user.is_staff)
    tag('date_joined', user.date_joined.isoformat())

    capture('user_signed_up', properties={'signup_method': 'form'})</pre>
</div>
{% endblock %}

fastapi>=0.109.0
uvicorn>=0.27.0
sqlalchemy>=2.0.0
python-dotenv>=1.0.0
posthog>=3.0.0
pydantic>=2.0.0
pydantic-settings>=2.0.0
jinja2>=3.0.0
python-multipart>=0.0.9
werkzeug>=3.0.0
itsdangerous>=2.0.0

"""Development server entry point."""

import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.main:app", host="0.0.0.0", port=5002, reload=True)