Python FastAPI 全栈集成

你将构建什么

本教程将带你从零构建一个 FastAPI 后端应用，实现袋袋 OAuth 登录、SQLAlchemy Token 持久化、按次计费事件上报、以及流式 AI 模型调用的完整流程。

前置条件

袋袋开发者账号

已注册袋袋账号并在开发者后台创建 App，获取 Client ID 和 Client Secret

开发环境

Python 3.10+、pip、基础 FastAPI 和 async/await 经验

Step 1: 项目初始化

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

创建 .env 文件存放凭证：

.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

永远不要将 .env 提交到版本控制。确保 .gitignore 包含该文件。

Step 2: 数据库配置

使用 SQLAlchemy async engine + SQLite（生产环境替换为 PostgreSQL）：

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

使用 SDK contrib 模块生成 Token 表：

models.py

from profy.contrib.sqlalchemy import create_token_model

ProfyOAuthToken = create_token_model(Base)

create_token_model 会自动创建一张包含 user_id、access_token、refresh_token、expires_at、scope 的表。无需手动定义 schema。

Step 3: 初始化 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 会在 SDK 自动刷新 Token 时回调，确保新 Token 立即落库，避免旧 Token 轮换后失效。

Step 4: OAuth 登录流程

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: 保护路由

通过 FastAPI 依赖注入自动加载并校验 Token：

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="未登录，请先授权")

    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

from fastapi import Depends

from dependencies import get_current_token


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

Step 6: 上报计费事件

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 保证同一请求重试不会重复扣费。建议在客户端生成后传入，或使用业务唯一标识。

Step 7: 调用 AI 模型

使用 httpx 直接调用袋袋的 OpenAI 兼容端点，实现 SSE 流式响应：

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

流式调用走 METERED 计费（按 token 用量），费用在流结束后自动结算。不需要手动调用 report_event。

Step 8: 错误处理中间件

SDK 抛出的异常均为类型化子类，可通过 FastAPI exception handler 统一拦截：

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": "登录已过期，请重新授权"},
    )


@app.exception_handler(InsufficientBalance)
async def handle_insufficient_balance(request: Request, exc: InsufficientBalance):
    return JSONResponse(
        status_code=402,
        content={"error": "余额不足，请充值后重试"},
    )


@app.exception_handler(InvalidEvent)
async def handle_invalid_event(request: Request, exc: InvalidEvent):
    return JSONResponse(
        status_code=400,
        content={"error": "事件名称无效，请检查 Meter 配置"},
    )


@app.exception_handler(ProfyApiError)
async def handle_profy_error(request: Request, exc: ProfyApiError):
    return JSONResponse(
        status_code=502,
        content={"error": f"袋袋服务异常: {exc}"},
    )

完整项目

三个核心文件的完整代码：

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="未登录，请先授权")

    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="袋袋 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": "已登录", "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": "登录已过期，请重新授权"})


@app.exception_handler(InsufficientBalance)
async def handle_insufficient_balance(request: Request, exc: InsufficientBalance):
    return JSONResponse(status_code=402, content={"error": "余额不足，请充值后重试"})


@app.exception_handler(InvalidEvent)
async def handle_invalid_event(request: Request, exc: InvalidEvent):
    return JSONResponse(status_code=400, content={"error": "事件名称无效，请检查 Meter 配置"})


@app.exception_handler(ProfyApiError)
async def handle_profy_error(request: Request, exc: ProfyApiError):
    return JSONResponse(status_code=502, content={"error": f"袋袋服务异常: {exc}"})

项目结构：

profy-fastapi-demo/
├── main.py               # FastAPI 应用入口
├── models.py             # SQLAlchemy 模型 + Token 表
├── config.py             # SDK 初始化 + Token Store
├── dependencies.py       # 路由依赖（Token 校验）
├── .env                  # 环境变量
└── requirements.txt

requirements.txt 内容：

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

启动开发服务器：

uvicorn main:app --reload --port 8000

部署建议

多 Worker 部署

生产环境使用 uvicorn main:app --workers 4 或搭配 Gunicorn：gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4。多 Worker 下 SQLite 不适用，需切换为 PostgreSQL。

环境变量管理

生产环境通过 Kubernetes Secret 或云平台环境变量注入凭证，不要使用 .env 文件。Cookie 需设置 secure=True。

数据库迁移

生产环境将 create_async_engine 的连接串替换为 postgresql+asyncpg://...，并使用 Alembic 管理 schema 迁移，不要在生产环境使用 create_all。

反向代理

在 Nginx / Caddy 后面运行，处理 HTTPS 终止和静态资源。确保 X-Forwarded-Proto 正确传递以生成准确的回调 URL。

下一步

Next.js 全栈集成

使用 TypeScript SDK 构建 Next.js 全栈应用

按次计费 SaaS

PER_USE 模式：固定价格按次扣费

Token 管理最佳实践

并发刷新、安全存储、降级策略

SDK 完整指南

双语言 API 详解与 Token 持久化

开始使用

认证与密钥

API 能力指南

SDK

集成

Cookbook

参考

Python FastAPI 全栈集成

你将构建什么

前置条件

袋袋开发者账号

开发环境

Step 1: 项目初始化

Step 2: 数据库配置

Step 3: 初始化 SDK

Step 4: OAuth 登录流程

Step 5: 保护路由

Step 6: 上报计费事件

Step 7: 调用 AI 模型

Step 8: 错误处理中间件

完整项目

部署建议

多 Worker 部署

环境变量管理

数据库迁移

反向代理

下一步

Next.js 全栈集成

按次计费 SaaS

Token 管理最佳实践

SDK 完整指南

​你将构建什么

​前置条件

袋袋开发者账号

开发环境

​Step 1: 项目初始化

​Step 2: 数据库配置

​Step 3: 初始化 SDK

​Step 4: OAuth 登录流程

​Step 5: 保护路由

​Step 6: 上报计费事件

​Step 7: 调用 AI 模型

​Step 8: 错误处理中间件

​完整项目

​部署建议

多 Worker 部署

环境变量管理

数据库迁移

反向代理

​下一步

Next.js 全栈集成

按次计费 SaaS

Token 管理最佳实践

SDK 完整指南

你将构建什么

前置条件

Step 1: 项目初始化

Step 2: 数据库配置

Step 3: 初始化 SDK

Step 4: OAuth 登录流程

Step 5: 保护路由

Step 6: 上报计费事件

Step 7: 调用 AI 模型

Step 8: 错误处理中间件

完整项目

部署建议

下一步