Back to Details

scaffolding-fastapi-dapr

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

FastAPI + Dapr Backend

FastAPI + Dapr 后端

Build production-grade FastAPI backends with SQLModel, Dapr integration, and JWT authentication.
使用SQLModel、Dapr集成和JWT认证构建生产级FastAPI后端。

Quick Start

快速开始

bash
undefined
bash
undefined

Project setup

项目初始化

uv init backend && cd backend uv add fastapi sqlmodel pydantic httpx python-jose uvicorn
uv init backend && cd backend uv add fastapi sqlmodel pydantic httpx python-jose uvicorn

Development

开发模式

uv run uvicorn main:app --reload --port 8000
uv run uvicorn main:app --reload --port 8000

With Dapr sidecar

搭配Dapr边车运行

dapr run --app-id myapp --app-port 8000 -- uvicorn main:app

---
dapr run --app-id myapp --app-port 8000 -- uvicorn main:app

---

FastAPI Core Patterns

FastAPI核心模式

1. SQLModel Schema (Database + API)

1. SQLModel 模型（数据库 + API）

python
from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional, Literal

class TaskBase(SQLModel):
    title: str = Field(max_length=200, index=True)
    status: Literal["pending", "in_progress", "completed"] = "pending"

class Task(TaskBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.now)

class TaskCreate(TaskBase):
    pass

class TaskRead(TaskBase):
    id: int
    created_at: datetime
python
from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional, Literal

class TaskBase(SQLModel):
    title: str = Field(max_length=200, index=True)
    status: Literal["pending", "in_progress", "completed"] = "pending"

class Task(TaskBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.now)

class TaskCreate(TaskBase):
    pass

class TaskRead(TaskBase):
    id: int
    created_at: datetime

2. Async Database Setup

2. 异步数据库配置

python
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
import os

DATABASE_URL = os.getenv("DATABASE_URL").replace("postgresql://", "postgresql+asyncpg://")
engine = create_async_engine(DATABASE_URL)

async def get_session() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session
python
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
import os

DATABASE_URL = os.getenv("DATABASE_URL").replace("postgresql://", "postgresql+asyncpg://")
engine = create_async_engine(DATABASE_URL)

async def get_session() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session

3. CRUD Endpoints

3. CRUD 端点

python
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import select

app = FastAPI()

@app.post("/tasks", response_model=TaskRead, status_code=201)
async def create_task(task: TaskCreate, session: AsyncSession = Depends(get_session)):
    db_task = Task.model_validate(task)
    session.add(db_task)
    await session.commit()
    await session.refresh(db_task)
    return db_task

@app.get("/tasks/{task_id}", response_model=TaskRead)
async def get_task(task_id: int, session: AsyncSession = Depends(get_session)):
    task = await session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Not found")
    return task

@app.patch("/tasks/{task_id}", response_model=TaskRead)
async def update_task(task_id: int, update: TaskUpdate, session: AsyncSession = Depends(get_session)):
    task = await session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Not found")
    update_data = update.model_dump(exclude_unset=True)
    task.sqlmodel_update(update_data)
    session.add(task)
    await session.commit()
    await session.refresh(task)
    return task
python
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import select

app = FastAPI()

@app.post("/tasks", response_model=TaskRead, status_code=201)
async def create_task(task: TaskCreate, session: AsyncSession = Depends(get_session)):
    db_task = Task.model_validate(task)
    session.add(db_task)
    await session.commit()
    await session.refresh(db_task)
    return db_task

@app.get("/tasks/{task_id}", response_model=TaskRead)
async def get_task(task_id: int, session: AsyncSession = Depends(get_session)):
    task = await session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="未找到")
    return task

@app.patch("/tasks/{task_id}", response_model=TaskRead)
async def update_task(task_id: int, update: TaskUpdate, session: AsyncSession = Depends(get_session)):
    task = await session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="未找到")
    update_data = update.model_dump(exclude_unset=True)
    task.sqlmodel_update(update_data)
    session.add(task)
    await session.commit()
    await session.refresh(task)
    return task

4. JWT/JWKS Authentication

4. JWT/JWKS 认证

python
from jose import jwt
import httpx

JWKS_URL = f"{SSO_URL}/.well-known/jwks.json"

async def get_current_user(authorization: str = Header()):
    token = authorization.replace("Bearer ", "")
    async with httpx.AsyncClient() as client:
        jwks = (await client.get(JWKS_URL)).json()
    payload = jwt.decode(token, jwks, algorithms=["RS256"])
    return payload

@app.get("/protected")
async def protected_route(user = Depends(get_current_user)):
    return {"user": user["sub"]}
See references/fastapi-patterns.md for audit logging, pagination, and OpenAPI configuration.
python
from jose import jwt
import httpx

JWKS_URL = f"{SSO_URL}/.well-known/jwks.json"

async def get_current_user(authorization: str = Header()):
    token = authorization.replace("Bearer ", "")
    async with httpx.AsyncClient() as client:
        jwks = (await client.get(JWKS_URL)).json()
    payload = jwt.decode(token, jwks, algorithms=["RS256"])
    return payload

@app.get("/protected")
async def protected_route(user = Depends(get_current_user)):
    return {"user": user["sub"]}
查看 references/fastapi-patterns.md 了解审计日志、分页和OpenAPI配置相关内容。

Dapr Integration Patterns

Dapr集成模式

1. Pub/Sub Subscription

1. 发布/订阅订阅配置

python
from fastapi import APIRouter, Request

router = APIRouter(prefix="/dapr", tags=["Dapr"])

@router.get("/subscribe")
async def subscribe():
    """Dapr calls this to discover subscriptions."""
    return [{
        "pubsubname": "pubsub",
        "topic": "task-created",
        "route": "/dapr/task-created"
    }]

@router.post("/task-created")
async def handle_task_created(request: Request, session: AsyncSession = Depends(get_session)):
    # CloudEvent wrapper - data is nested
    event = await request.json()
    task_data = event.get("data", event)  # Handle both wrapped and unwrapped

    # Process event
    task = Task.model_validate(task_data)
    session.add(task)
    await session.commit()
    return {"status": "processed"}
python
from fastapi import APIRouter, Request

router = APIRouter(prefix="/dapr", tags=["Dapr"])

@router.get("/subscribe")
async def subscribe():
    """Dapr会调用此接口发现订阅配置。"""
    return [{
        "pubsubname": "pubsub",
        "topic": "task-created",
        "route": "/dapr/task-created"
    }]

@router.post("/task-created")
async def handle_task_created(request: Request, session: AsyncSession = Depends(get_session)):
    # CloudEvent 包装器 - 数据嵌套在内
    event = await request.json()
    task_data = event.get("data", event)  # 兼容包装和未包装两种格式

    # 处理事件
    task = Task.model_validate(task_data)
    session.add(task)
    await session.commit()
    return {"status": "processed"}

2. Publishing Events

2. 发布事件

python
import httpx

DAPR_URL = "http://localhost:3500"

async def publish_event(topic: str, data: dict):
    async with httpx.AsyncClient() as client:
        await client.post(
            f"{DAPR_URL}/v1.0/publish/pubsub/{topic}",
            json=data,
            headers={"Content-Type": "application/json"}
        )
python
import httpx

DAPR_URL = "http://localhost:3500"

async def publish_event(topic: str, data: dict):
    async with httpx.AsyncClient() as client:
        await client.post(
            f"{DAPR_URL}/v1.0/publish/pubsub/{topic}",
            json=data,
            headers={"Content-Type": "application/json"}
        )

3. Scheduled Jobs

3. 调度任务

python
undefined
python
undefined

Schedule a job via Dapr Jobs API (alpha)

通过Dapr Jobs API（alpha版本）调度任务

async def schedule_job(name: str, schedule: str, callback_url: str, data: dict): async with httpx.AsyncClient() as client: await client.post( f"{DAPR_URL}/v1.0-alpha1/jobs/{name}", json={ "schedule": schedule, # "@every 5m" or "0 */5 * * * *" "data": data, }, headers={"dapr-app-callback-url": callback_url} )
async def schedule_job(name: str, schedule: str, callback_url: str, data: dict): async with httpx.AsyncClient() as client: await client.post( f"{DAPR_URL}/v1.0-alpha1/jobs/{name}", json={ "schedule": schedule, # 格式如 "@every 5m" 或 "0 */5 * * * *" "data": data, }, headers={"dapr-app-callback-url": callback_url} )

Job callback endpoint

任务回调端点

@app.post("/jobs/process") async def process_job(request: Request): job_data = await request.json() # Handle job execution return {"status": "completed"}

See [references/dapr-patterns.md](references/dapr-patterns.md) for state management and advanced patterns.

---
@app.post("/jobs/process") async def process_job(request: Request): job_data = await request.json() # 执行任务处理逻辑 return {"status": "completed"}

查看 [references/dapr-patterns.md](references/dapr-patterns.md) 了解状态管理和高级模式。

---

Production Patterns

生产环境模式

Structured Logging

结构化日志

python
import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ]
)
log = structlog.get_logger()
log.info("task_created", task_id=task.id, user_id=user["sub"])
python
import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ]
)
log = structlog.get_logger()
log.info("task_created", task_id=task.id, user_id=user["sub"])

Repository + Service Pattern

仓库 + 服务模式

python
undefined
python
undefined

Repository: data access only

仓库层：仅负责数据访问

class TaskRepository: def init(self, session: AsyncSession): self.session = session
async def create(self, task: TaskCreate) -> Task:
    db_task = Task.model_validate(task)
    self.session.add(db_task)
    await self.session.commit()
    return db_task
class TaskRepository: def init(self, session: AsyncSession): self.session = session
async def create(self, task: TaskCreate) -> Task:
    db_task = Task.model_validate(task)
    self.session.add(db_task)
    await self.session.commit()
    return db_task

Service: business logic

服务层：处理业务逻辑

class TaskService: def init(self, repo: TaskRepository): self.repo = repo
async def create_task(self, task: TaskCreate, user_id: str) -> Task:
    # Business logic here
    return await self.repo.create(task)
class TaskService: def init(self, repo: TaskRepository): self.repo = repo
async def create_task(self, task: TaskCreate, user_id: str) -> Task:
    # 此处添加业务逻辑
    return await self.repo.create(task)

Dependency injection

依赖注入

def get_task_service(session: AsyncSession = Depends(get_session)): return TaskService(TaskRepository(session))
undefined
def get_task_service(session: AsyncSession = Depends(get_session)): return TaskService(TaskRepository(session))
undefined

Async Testing

异步测试

python
@pytest.fixture
async def client(session):
    app.dependency_overrides[get_session] = lambda: session
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

@pytest.mark.anyio
async def test_create_task(client: AsyncClient):
    response = await client.post("/tasks", json={"title": "Test"})
    assert response.status_code == 201
See references/production-testing.md for full patterns.
python
@pytest.fixture
async def client(session):
    app.dependency_overrides[get_session] = lambda: session
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

@pytest.mark.anyio
async def test_create_task(client: AsyncClient):
    response = await client.post("/tasks", json={"title": "Test"})
    assert response.status_code == 201
查看 references/production-testing.md 了解完整模式。

Project Structure

项目结构

backend/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI app
│   ├── database.py       # Async engine + session
│   ├── models/           # SQLModel schemas
│   ├── routers/          # API routes
│   ├── repositories/     # Data access layer
│   ├── services/         # Business logic
│   └── dapr/             # Dapr handlers
├── tests/
│   ├── conftest.py       # Fixtures
│   └── test_*.py         # Test files
├── components/           # Dapr components (k8s)
│   ├── pubsub.yaml
│   └── statestore.yaml
└── pyproject.toml
backend/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI 应用
│   ├── database.py       # 异步引擎 + 会话
│   ├── models/           # SQLModel 模型
│   ├── routers/          # API 路由
│   ├── repositories/     # 数据访问层
│   ├── services/         # 业务逻辑层
│   └── dapr/             # Dapr 处理器
├── tests/
│   ├── conftest.py       # 测试夹具
│   └── test_*.py         # 测试文件
├── components/           # Dapr 组件（K8s）
│   ├── pubsub.yaml
│   └── statestore.yaml
└── pyproject.toml

Verification

验证

Run: 
python3 scripts/verify.py
Expected: 
✓ scaffolding-fastapi-dapr skill ready
运行：
python3 scripts/verify.py
预期结果：
✓ scaffolding-fastapi-dapr skill ready

If Verification Fails

验证失败时的处理

  1. Check: references/ folder has both pattern files
  2. Stop and report if still failing
  1. 检查：references/目录下是否包含所有模式文件
  2. 如果仍失败，请停止操作并上报

Related Skills

相关技能

  • configuring-better-auth - JWT/JWKS auth for API endpoints
  • fetching-library-docs - FastAPI docs: 
    --library-id /fastapi/fastapi --topic dependencies
  • configuring-better-auth - 用于API端点的JWT/JWKS认证
  • fetching-library-docs - FastAPI文档：
    --library-id /fastapi/fastapi --topic dependencies

References

参考资料

  • references/fastapi-patterns.md - Complete FastAPI backend patterns
  • references/dapr-patterns.md - Dapr pub/sub, state, and jobs
  • references/sqlmodel-patterns.md - SQLModel database patterns and migrations
  • references/production-testing.md - Structured logging, DI, testing, versioning
  • references/fastapi-patterns.md - 完整的FastAPI后端模式
  • references/dapr-patterns.md - Dapr发布/订阅、状态管理和任务调度
  • references/sqlmodel-patterns.md - SQLModel数据库模式和迁移
  • references/production-testing.md - 结构化日志、依赖注入、测试、版本控制