> ## Documentation Index
> Fetch the complete documentation index at: https://docs.profy.cn/llms.txt
> Use this file to discover all available pages before exploring further.

# Python FastAPI Full-Stack Integration

> Build a complete FastAPI application using the Profy Python SDK, covering OAuth login, SQLAlchemy Token persistence, billing event reporting, and streaming AI calls

## What You'll Build

This tutorial walks you through building a FastAPI backend application from scratch, implementing Profy OAuth login, SQLAlchemy Token persistence, per-use billing event reporting, and streaming AI model calls.

```mermaid theme={null}
sequenceDiagram
    participant User as User Browser
    participant App as FastAPI App
    participant Profy as Profy OAuth

    User->>App: Click "Login with Profy"
    App->>Profy: Redirect to /oauth/authorize
    Profy->>User: Display authorization page
    User->>Profy: Confirm authorization
    Profy->>App: Callback ?code=xxx
    App->>Profy: exchange_code(code)
    Profy->>App: TokenData(access_token, refresh_token, ...)
    App->>App: Store Token in DB
    User->>App: Trigger paid action
    App->>Profy: report_event(event_name, token)
    Profy->>App: { charged, balance_remaining }
```

## Prerequisites

<CardGroup cols={2}>
  <Card title="Profy Developer Account" icon="user">
    Registered a Profy account and created an App in the developer dashboard to obtain your Client ID and Client Secret
  </Card>

  <Card title="Development Environment" icon="code">
    Python 3.10+, pip, basic FastAPI and async/await experience
  </Card>
</CardGroup>

## Step 1: Project Setup

```bash theme={null}
mkdir profy-fastapi-demo && cd profy-fastapi-demo
python -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn profy-sdk[sqlalchemy] sqlalchemy aiosqlite httpx python-dotenv
```

Create a `.env` file for credentials:

```env .env theme={null}
PROFY_CLIENT_ID=your_app_client_id
PROFY_CLIENT_SECRET=your_app_client_secret
PROFY_CALLBACK_URL=http://localhost:8000/callback
PROFY_BASE_URL=https://app.profy.cn
```

<Warning>
  Never commit `.env` to version control. Make sure `.gitignore` includes this file.
</Warning>

## Step 2: Database Configuration

Use SQLAlchemy async engine + SQLite (replace with PostgreSQL for production):

```python models.py theme={null}
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

engine = create_async_engine("sqlite+aiosqlite:///./profy_demo.db")
async_session = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass
```

Use the SDK contrib module to generate the Token table:

```python models.py theme={null}
from profy.contrib.sqlalchemy import create_token_model

ProfyOAuthToken = create_token_model(Base)
```

<Note>
  `create_token_model` automatically creates a table with `user_id`, `access_token`, `refresh_token`, `expires_at`, and `scope` columns. No manual schema definition needed.
</Note>

## Step 3: Initialize the SDK

```python config.py theme={null}
import os

from dotenv import load_dotenv
from profy import ProfyApp
from profy.contrib.sqlalchemy import SQLAlchemyTokenStore

from models import ProfyOAuthToken, async_session

load_dotenv()

store = SQLAlchemyTokenStore(async_session, ProfyOAuthToken)


async def _on_token_refresh(user_id: str, new_token):
    await store.save(user_id, new_token, scope="events:write")


profy = ProfyApp(
    client_id=os.environ["PROFY_CLIENT_ID"],
    client_secret=os.environ["PROFY_CLIENT_SECRET"],
    base_url=os.environ["PROFY_BASE_URL"],
    on_token_refresh=_on_token_refresh,
)

PROFY_AUTHORIZE_URL = f"{os.environ['PROFY_BASE_URL']}/oauth/authorize"
PROFY_CALLBACK_URL = os.environ["PROFY_CALLBACK_URL"]
```

<Tip>
  `on_token_refresh` is called when the SDK automatically refreshes a Token, ensuring the new Token is persisted immediately and preventing the old Token from being invalidated with the new one lost.
</Tip>

## Step 4: OAuth Login Flow

```python main.py theme={null}
import os
from contextlib import asynccontextmanager
from urllib.parse import urlencode

from fastapi import FastAPI
from fastapi.responses import RedirectResponse

from config import PROFY_AUTHORIZE_URL, PROFY_CALLBACK_URL, profy, store
from models import Base, engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(lifespan=lifespan)


@app.get("/login")
async def login():
    params = urlencode({
        "client_id": os.environ["PROFY_CLIENT_ID"],
        "redirect_uri": PROFY_CALLBACK_URL,
        "scope": "events:write",
        "response_type": "code",
    })
    return RedirectResponse(f"{PROFY_AUTHORIZE_URL}?{params}")


@app.get("/callback")
async def callback(code: str):
    token = await profy.exchange_code(code, PROFY_CALLBACK_URL)
    await store.save(token.user_id, token, scope="events:write")
    response = RedirectResponse("/dashboard")
    response.set_cookie(
        "profy_user_id",
        token.user_id,
        httponly=True,
        samesite="lax",
        max_age=60 * 60 * 24 * 90,
    )
    return response
```

## Step 5: Protecting Routes

Use FastAPI dependency injection to automatically load and validate Tokens:

```python dependencies.py theme={null}
from fastapi import Cookie, HTTPException
from profy.contrib.sqlalchemy import SQLAlchemyTokenStore

from config import profy, store


async def get_current_token(profy_user_id: str = Cookie()):
    saved = await store.load(profy_user_id)
    if saved is None:
        raise HTTPException(status_code=401, detail="Not logged in. Please authorize first.")

    if saved.is_expired:
        refreshed = await profy.refresh_token(saved.refresh_token)
        await store.save(profy_user_id, refreshed, scope="events:write")
        return refreshed

    return saved
```

Inject in routes:

```python main.py theme={null}
from fastapi import Depends

from dependencies import get_current_token


@app.get("/dashboard")
async def dashboard(token=Depends(get_current_token)):
    return {"message": "Logged in", "expires_at": token.expires_at}
```

## Step 6: Report Billing Events

```python main.py theme={null}
import uuid

from pydantic import BaseModel


class GenerateRequest(BaseModel):
    prompt: str


@app.post("/api/generate")
async def generate(body: GenerateRequest, token=Depends(get_current_token)):
    result = await profy.report_event(
        "ai_generation",
        token=token,
        idempotency_key=str(uuid.uuid4()),
        metadata={"prompt_length": len(body.prompt)},
    )
    return {
        "charged": result.charged,
        "balance_remaining": result.balance_remaining,
        "output": f"Generated content for: {body.prompt}",
    }
```

<Note>
  `idempotency_key` ensures the same request won't be charged twice on retry. It's recommended to generate it on the client side or use a business-unique identifier.
</Note>

## Step 7: Call AI Models

Use `httpx` to directly call Profy's OpenAI-compatible endpoint with SSE streaming responses:

```python main.py theme={null}
import httpx
from fastapi.responses import StreamingResponse


class ChatRequest(BaseModel):
    message: str


@app.post("/api/chat")
async def chat(body: ChatRequest, token=Depends(get_current_token)):
    base_url = os.environ["PROFY_BASE_URL"]

    async def stream_response():
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                f"{base_url}/openapi/v1/events/chat",
                headers={
                    "Authorization": f"Bearer {token.access_token}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": "deepseek-v3",
                    "messages": [{"role": "user", "content": body.message}],
                    "stream": True,
                },
                timeout=60.0,
            ) as resp:
                resp.raise_for_status()
                async for line in resp.aiter_lines():
                    if line.startswith("data: "):
                        yield f"{line}\n\n"

    return StreamingResponse(stream_response(), media_type="text/event-stream")
```

<Warning>
  Streaming calls use METERED billing (per-token usage). Charges are settled automatically when the stream ends. No manual `report_event` call is needed.
</Warning>

## Step 8: Error Handling Middleware

All SDK exceptions are typed subclasses that can be intercepted uniformly via FastAPI exception handlers:

```python main.py theme={null}
from fastapi import Request
from fastapi.responses import JSONResponse
from profy import AuthExpired, InsufficientBalance, InvalidEvent, ProfyApiError


@app.exception_handler(AuthExpired)
async def handle_auth_expired(request: Request, exc: AuthExpired):
    return JSONResponse(
        status_code=401,
        content={"error": "Session expired. Please re-authorize."},
    )


@app.exception_handler(InsufficientBalance)
async def handle_insufficient_balance(request: Request, exc: InsufficientBalance):
    return JSONResponse(
        status_code=402,
        content={"error": "Insufficient balance. Please top up and try again."},
    )


@app.exception_handler(InvalidEvent)
async def handle_invalid_event(request: Request, exc: InvalidEvent):
    return JSONResponse(
        status_code=400,
        content={"error": "Invalid event name. Please check your Meter configuration."},
    )


@app.exception_handler(ProfyApiError)
async def handle_profy_error(request: Request, exc: ProfyApiError):
    return JSONResponse(
        status_code=502,
        content={"error": f"Profy service error: {exc}"},
    )
```

## Complete Project

Full code for the three core files:

```python models.py theme={null}
from profy.contrib.sqlalchemy import create_token_model
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

engine = create_async_engine("sqlite+aiosqlite:///./profy_demo.db")
async_session = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


ProfyOAuthToken = create_token_model(Base)
```

```python config.py theme={null}
import os

from dotenv import load_dotenv
from profy import ProfyApp
from profy.contrib.sqlalchemy import SQLAlchemyTokenStore

from models import ProfyOAuthToken, async_session

load_dotenv()

store = SQLAlchemyTokenStore(async_session, ProfyOAuthToken)


async def _on_token_refresh(user_id: str, new_token):
    await store.save(user_id, new_token, scope="events:write")


profy = ProfyApp(
    client_id=os.environ["PROFY_CLIENT_ID"],
    client_secret=os.environ["PROFY_CLIENT_SECRET"],
    base_url=os.environ["PROFY_BASE_URL"],
    on_token_refresh=_on_token_refresh,
)

PROFY_AUTHORIZE_URL = f"{os.environ['PROFY_BASE_URL']}/oauth/authorize"
PROFY_CALLBACK_URL = os.environ["PROFY_CALLBACK_URL"]
```

```python dependencies.py theme={null}
from fastapi import Cookie, HTTPException

from config import profy, store


async def get_current_token(profy_user_id: str = Cookie()):
    saved = await store.load(profy_user_id)
    if saved is None:
        raise HTTPException(status_code=401, detail="Not logged in. Please authorize first.")

    if saved.is_expired:
        refreshed = await profy.refresh_token(saved.refresh_token)
        await store.save(profy_user_id, refreshed, scope="events:write")
        return refreshed

    return saved
```

```python main.py theme={null}
import os
import uuid
from contextlib import asynccontextmanager
from urllib.parse import urlencode

import httpx
from fastapi import Depends, FastAPI, Request
from fastapi.responses import JSONResponse, RedirectResponse, StreamingResponse
from profy import AuthExpired, InsufficientBalance, InvalidEvent, ProfyApiError
from pydantic import BaseModel

from config import PROFY_AUTHORIZE_URL, PROFY_CALLBACK_URL, profy, store
from dependencies import get_current_token
from models import Base, engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(title="Profy FastAPI Demo", lifespan=lifespan)


# --- OAuth ---


@app.get("/login")
async def login():
    params = urlencode({
        "client_id": os.environ["PROFY_CLIENT_ID"],
        "redirect_uri": PROFY_CALLBACK_URL,
        "scope": "events:write",
        "response_type": "code",
    })
    return RedirectResponse(f"{PROFY_AUTHORIZE_URL}?{params}")


@app.get("/callback")
async def callback(code: str):
    token = await profy.exchange_code(code, PROFY_CALLBACK_URL)
    await store.save(token.user_id, token, scope="events:write")
    response = RedirectResponse("/dashboard")
    response.set_cookie(
        "profy_user_id",
        token.user_id,
        httponly=True,
        samesite="lax",
        max_age=60 * 60 * 24 * 90,
    )
    return response


# --- Protected routes ---


@app.get("/dashboard")
async def dashboard(token=Depends(get_current_token)):
    return {"message": "Logged in", "expires_at": token.expires_at}


class GenerateRequest(BaseModel):
    prompt: str


@app.post("/api/generate")
async def generate(body: GenerateRequest, token=Depends(get_current_token)):
    result = await profy.report_event(
        "ai_generation",
        token=token,
        idempotency_key=str(uuid.uuid4()),
        metadata={"prompt_length": len(body.prompt)},
    )
    return {
        "charged": result.charged,
        "balance_remaining": result.balance_remaining,
        "output": f"Generated content for: {body.prompt}",
    }


class ChatRequest(BaseModel):
    message: str


@app.post("/api/chat")
async def chat(body: ChatRequest, token=Depends(get_current_token)):
    base_url = os.environ["PROFY_BASE_URL"]

    async def stream_response():
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                f"{base_url}/openapi/v1/events/chat",
                headers={
                    "Authorization": f"Bearer {token.access_token}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": "deepseek-v3",
                    "messages": [{"role": "user", "content": body.message}],
                    "stream": True,
                },
                timeout=60.0,
            ) as resp:
                resp.raise_for_status()
                async for line in resp.aiter_lines():
                    if line.startswith("data: "):
                        yield f"{line}\n\n"

    return StreamingResponse(stream_response(), media_type="text/event-stream")


# --- Error handlers ---


@app.exception_handler(AuthExpired)
async def handle_auth_expired(request: Request, exc: AuthExpired):
    return JSONResponse(status_code=401, content={"error": "Session expired. Please re-authorize."})


@app.exception_handler(InsufficientBalance)
async def handle_insufficient_balance(request: Request, exc: InsufficientBalance):
    return JSONResponse(status_code=402, content={"error": "Insufficient balance. Please top up and try again."})


@app.exception_handler(InvalidEvent)
async def handle_invalid_event(request: Request, exc: InvalidEvent):
    return JSONResponse(status_code=400, content={"error": "Invalid event name. Please check your Meter configuration."})


@app.exception_handler(ProfyApiError)
async def handle_profy_error(request: Request, exc: ProfyApiError):
    return JSONResponse(status_code=502, content={"error": f"Profy service error: {exc}"})
```

Project structure:

```
profy-fastapi-demo/
├── main.py               # FastAPI application entry point
├── models.py             # SQLAlchemy models + Token table
├── config.py             # SDK initialization + Token Store
├── dependencies.py       # Route dependencies (Token validation)
├── .env                  # Environment variables
└── requirements.txt
```

`requirements.txt` contents:

```txt requirements.txt theme={null}
fastapi>=0.115
uvicorn>=0.34
profy-sdk[sqlalchemy]>=0.1
sqlalchemy>=2.0
aiosqlite>=0.20
httpx>=0.28
python-dotenv>=1.0
```

Start the development server:

```bash theme={null}
uvicorn main:app --reload --port 8000
```

## Deployment Recommendations

<CardGroup cols={2}>
  <Card title="Multi-Worker Deployment" icon="server">
    Use `uvicorn main:app --workers 4` or pair with Gunicorn: `gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4` for production. SQLite is not suitable for multi-worker setups — switch to PostgreSQL.
  </Card>

  <Card title="Environment Variable Management" icon="key">
    Inject credentials via Kubernetes Secrets or cloud platform environment variables in production — don't use `.env` files. Set `secure=True` for cookies.
  </Card>

  <Card title="Database Migrations" icon="database">
    Replace the `create_async_engine` connection string with `postgresql+asyncpg://...` for production. Use Alembic for schema migrations — don't use `create_all` in production.
  </Card>

  <Card title="Reverse Proxy" icon="shield">
    Run behind Nginx / Caddy to handle HTTPS termination and static assets. Ensure `X-Forwarded-Proto` is correctly passed to generate accurate callback URLs.
  </Card>
</CardGroup>

## Next Steps

<CardGroup cols={2}>
  <Card title="Next.js Full-Stack Integration" icon="react" href="/en/developers/cookbook/nextjs-fullstack">
    Build a Next.js full-stack application with the TypeScript SDK
  </Card>

  <Card title="Per-Use Billing SaaS" icon="coins" href="/en/developers/cookbook/per-use-billing">
    PER\_USE mode: fixed-price per-use billing
  </Card>

  <Card title="Token Management Best Practices" icon="shield-check" href="/en/developers/cookbook/token-management">
    Concurrent refresh, secure storage, degradation strategies
  </Card>

  <Card title="SDK Complete Guide" icon="book" href="/en/developers/sdk-python">
    Dual-language API reference and Token persistence
  </Card>
</CardGroup>