Python FastAPI Full-Stack Integration

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.

Prerequisites

Profy Developer Account

Registered a Profy account and created an App in the developer dashboard to obtain your Client ID and Client Secret

Development Environment

Python 3.10+, pip, basic FastAPI and async/await experience

Step 1: Project Setup

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

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

Never commit .env to version control. Make sure .gitignore includes this file.

Step 2: Database Configuration

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

models.py

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:

models.py

from profy.contrib.sqlalchemy import create_token_model

ProfyOAuthToken = create_token_model(Base)

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

Step 3: Initialize the SDK

config.py

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"]

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.

main.py

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:

dependencies.py

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:

main.py

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

main.py

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}",
    }

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.

Step 7: Call AI Models

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

main.py

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")

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

Step 8: Error Handling Middleware

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

main.py

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:

models.py

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)

config.py

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"]

dependencies.py

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

main.py

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:

requirements.txt

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:

uvicorn main:app --reload --port 8000

Deployment Recommendations

Multi-Worker Deployment

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.

Environment Variable Management

Inject credentials via Kubernetes Secrets or cloud platform environment variables in production — don’t use .env files. Set secure=True for cookies.

Database Migrations

Replace the create_async_engine connection string with postgresql+asyncpg://... for production. Use Alembic for schema migrations — don’t use create_all in production.

Reverse Proxy

Run behind Nginx / Caddy to handle HTTPS termination and static assets. Ensure X-Forwarded-Proto is correctly passed to generate accurate callback URLs.

Next Steps

Next.js Full-Stack Integration

Build a Next.js full-stack application with the TypeScript SDK

Per-Use Billing SaaS

PER_USE mode: fixed-price per-use billing

Token Management Best Practices

Concurrent refresh, secure storage, degradation strategies

SDK Complete Guide

Dual-language API reference and Token persistence

Get Started

Authentication & Keys

API Guides

SDK

Integrations

Cookbook

Reference

Python FastAPI Full-Stack Integration

What You’ll Build

Prerequisites

Profy Developer Account

Development Environment

Step 1: Project Setup

Step 2: Database Configuration

Step 3: Initialize the SDK

Step 5: Protecting Routes

Step 6: Report Billing Events

Step 7: Call AI Models

Step 8: Error Handling Middleware

Complete Project

Deployment Recommendations

Multi-Worker Deployment

Environment Variable Management

Database Migrations

Reverse Proxy

Next Steps

Next.js Full-Stack Integration

Per-Use Billing SaaS

Token Management Best Practices

SDK Complete Guide

​What You’ll Build

​Prerequisites

Profy Developer Account

Development Environment

​Step 1: Project Setup

​Step 2: Database Configuration

​Step 3: Initialize the SDK

​Step 4: OAuth Login Flow

​Step 5: Protecting Routes

​Step 6: Report Billing Events

​Step 7: Call AI Models

​Step 8: Error Handling Middleware

​Complete Project

​Deployment Recommendations

Multi-Worker Deployment

Environment Variable Management

Database Migrations

Reverse Proxy

​Next Steps

Next.js Full-Stack Integration

Per-Use Billing SaaS

Token Management Best Practices

SDK Complete Guide

What You’ll Build

Prerequisites

Step 1: Project Setup

Step 2: Database Configuration

Step 3: Initialize the SDK

Step 4: OAuth Login Flow

Step 5: Protecting Routes

Step 6: Report Billing Events

Step 7: Call AI Models

Step 8: Error Handling Middleware

Complete Project

Deployment Recommendations

Next Steps