Python FastAPI 开发 使用 uv 包管理器、模块化项目结构、SQLAlchemy ORM 和生产就绪模式构建 Python API 的专家模式。
技术栈
- 运行时: Python 3.12+
- 包管理器: uv（快速，基于 Rust）
- 框架: FastAPI
- ORM: SQLAlchemy 2.0（异步）
- 验证: Pydantic v2
- 数据库: PostgreSQL（或 SQLite 用于开发）
- 迁移: Alembic
- 测试: pytest, pytest-asyncio
- 代码检查: ruff
项目结构
按领域组织的特征模块化架构 - 代码按领域组织，而不是按层：
我的项目/
├── pyproject.toml # 带有 uv 的项目配置
├── uv.lock # 锁定文件
├── .python-version # Python 版本
├── .env # 环境变量
├── .env.example
├── alembic.ini # Alembic 配置
├── alembic/ # 迁移
│ ├── env.py
│ ├── script.py.mako
│ └── versions/
├── src/
│ └── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 应用入口
│ ├── config.py # 设置
│ ├── database.py # 数据库会话
│ ├── core/
│ │ ├── __init__.py
│ │ ├── dependencies.py # 共享依赖项
│ │ ├── exceptions.py # 自定义异常
│ │ ├── middleware.py # 中间件
│ │ └── security.py # 认证工具
│ ├── models/
│ │ ├── __init__.py
│ │ └── base.py # SQLAlchemy 基础和混合
│ ├── features/
│ │ ├── __init__.py
│ │ ├── auth/
│ │ │ ├── __init__.py
│ │ │ ├── api.py # 认证端点
│ │ │ ├── schemas.py # 认证 Pydantic 模式
│ │ │ ├── services.py # 认证业务逻辑
│ │ │ ├── models.py # 认证 SQLAlchemy 模型
│ │ │ └── utils.py # 认证助手（JWT 等）
│ │ ├── users/
│ │ │ ├── __init__.py
│ │ │ ├── api.py # 用户端点
│ │ │ ├── schemas.py # 用户 Pydantic 模式
│ │ │ ├── services.py # 用户业务逻辑
│ │ │ ├── models.py # 用户 SQLAlchemy 模型
│ │ │ └── repository.py # 用户数据访问
│ │ └── items/
│ │ ├── __init__.py
│ │ ├── api.py
│ │ ├── schemas.py
│ │ ├── services.py
│ │ └── models.py
│ └── api/
│ ├── __init__.py
│ └── router.py # 聚合所有功能路由器
└── tests/
├── __init__.py
├── conftest.py
├── features/
│ ├── auth/
│ │ └── test_auth.py
│ └── users/
│ └── test_users.py
└── integration/
使用 uv 快速设置
# 安装 uv
curl -LsSf https://astral.sh/uv/install.sh | sh
# 创建新项目
uv init my-project
cd my-project
# 设置 Python 版本
uv python pin 3.12
# 添加依赖项
uv add fastapi uvicorn[standard] sqlalchemy[asyncio] asyncpg
uv add pydantic pydantic-settings python-dotenv
uv add alembic
# 添加开发依赖项
uv add --dev pytest pytest-asyncio pytest-cov httpx ruff mypy
# 创建源结构
mkdir -p src/app/{api/v1/endpoints,core,models,schemas,services,repositories}
touch src/app/__init__.py
核心模式
pyproject.toml
[项目]
名称 = "my-project"
版本 = "0.1.0"
描述 = "FastAPI 应用程序"
需要 Python = ">=3.12"
依赖 = [
"fastapi>=0.115.0",
"uvicorn[standard]>=0.32.0",
"sqlalchemy[asyncio]>=2.0.0",
"asyncpg>=0.30.0",
"pydantic>=2.10.0",
"pydantic-settings>=2.6.0",
"python-dotenv>=1.0.0",
"alembic>=1.14.0",
]
[项目.可选依赖]
开发 = [
"pytest>=8.0.0",
"pytest-asyncio>=0.24.0",
"pytest-cov>=6.0.0",
"httpx>=0.28.0",
"ruff>=0.8.0",
"mypy>=1.13.0",
]
[工具.ruff]
目标版本 = "py312"
行长度 = 88
[工具.ruff.lint]
选择 = ["E", "F", "I", "N", "W", "UP", "B", "C4", "SIM"]
[工具.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
[工具.mypy]
python_version = "3.12"
严格 = true
配置（src/app/config.py）
从 functools import lru_cache
从 pydantic_settings import BaseSettings, SettingsConfigDict
类 Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
)
# App
app_name: str = "My API"
debug: bool = False
api_v1_prefix: str = "/api/v1"
# 数据库
database_url: str = "postgresql+asyncpg://user:pass@localhost:5432/db"
# 安全
secret_key: str = "change-me-in-production"
access_token_expire_minutes: int = 30
@lru_cache
def get_settings() -> Settings:
return Settings()
设置 = get_settings()
数据库设置（src/app/database.py）
从 collections.abc import AsyncGenerator
从 sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
create_async_engine,
)
从 app.config import 设置
引擎 = create_async_engine(
设置.database_url,
echo=设置.debug,
pool_pre_ping=True,
)
async_session_maker = async_sessionmaker(
引擎,
class_=AsyncSession,
expire_on_commit=False,
)
异步 def get_db() -> AsyncGenerator[AsyncSession, None]:
异步与 async_session_maker() 作为 session:
尝试:
产量 session
等待 session.commit()
除了 Exception:
等待 session.rollback()
引发
SQLAlchemy 基础模型（src/app/models/base.py）
从 datetime import datetime
从 sqlalchemy import DateTime, func
从 sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
类 Base(DeclarativeBase):
通过
类 TimestampMixin:
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
服务器默认=func.now(),
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
服务器默认=func.now(),
onupdate=func.now(),
)
功能模块模式
每个功能都是自包含的，有自己的 api、模式、服务、模型和工具。
功能：users/schemas.py
从 pydantic import BaseModel, EmailStr, ConfigDict
类 UserBase(BaseModel):
email: EmailStr
full_name: str | None = None
类 UserCreate(UserBase):
password: str
类 UserUpdate(BaseModel):
email: EmailStr | None = None
full_name: str | None = None
password: str | None = None
类 UserResponse(UserBase):
model_config = ConfigDict(from_attributes=True)
id: int
is_active: bool
功能：users/models.py
从 sqlalchemy import String
从 sqlalchemy.orm import Mapped, mapped_column
从 app.models.base import Base, TimestampMixin
类 User(Base, TimestampMixin):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
hashed_password: Mapped[str] = mapped_column(String(255))
full_name: Mapped[str | None] = mapped_column(String(255))
is_active: Mapped[bool] = mapped_column默认=True)
功能：users/repository.py
从 sqlalchemy import select
从 sqlalchemy.ext.asyncio import AsyncSession
从 app.features.users.models import User
类 UserRepository:
防御 __init__(self, db: AsyncSession):
自我.db = db
异步 def get(self, id: int) -> User | None:
结果 = 等待 self.db.execute(select(User).where(User.id == id))
返回结果.scalar_one_or_none()
异步 def get_by_email(self, email: str) -> User | None:
结果 = 等待 self.db.execute(select(User).where(User.email == email))
返回结果.scalar_one_or_none()
异步 def get_all(self, skip: int = 0, limit: int = 100) -> 列表[User]:
结果 = 等待 self.db.execute(select(User).offset(skip).limit(limit))
返回列表结果.scalars().all()
异步 def create(self, data: 字典) -> User:
用户 = User(**数据)
自我.db.add(user)
等待 self.db.flush()
等待 self.db.refresh(user)
返回用户
异步 def update(self, 用户: User, 数据: 字典) -> User:
对于字段，值 在 数据.items():
如果值不是 None:
设置attr(user, field, value)
等待 self.db.flush()
等待 self.db.refresh(user)
返回用户
功能：users/services.py
从 sqlalchemy.ext.asyncio import AsyncSession
从 app.core.security import hash_password
从 app.features.users.models import User
从 app.features.users.repository import UserRepository
从 app.features.users.schemas import UserCreate, UserUpdate
类 UserService:
防御 __init__(self, db: AsyncSession):
自我.db = db
自我.repo = UserRepository(db)
异步 def get(self, user_id: int) -> User | None:
返回等待 self.repo.get(user_id)
异步 def get_by_email(self, email: str) -> User | None:
返回等待 self.repo.get_by_email(email)
异步 def list(self, skip: int = 0, limit: int = 100) -> 列表[User]:
返回等待 self.repo.get_all(skip=skip, limit=limit)
异步 def create(self, user_in: UserCreate) -> User:
数据 = user_in.model_dump()
数据["hashed_password"] = hash_password(data.pop("password"))
返回等待 self.repo.create(data)
异步 def update(self, 用户: User, user_in: UserUpdate) -> User:
数据 = user_in.model_dump(exclude_unset=True)
如果 "password" 在数据:
数据["hashed_password"] = hash_password(data.pop("password"))
返回等待 self.repo.update(user, data)
功能：users/api.py
从 fastapi import APIRouter, Depends, HTTPException, 状态
从 sqlalchemy.ext.asyncio import AsyncSession
从 app.database import get_db
从 app.features.users.schemas import UserCreate, UserResponse, UserUpdate
从 app.features.users.services import UserService
路由器 = APIRouter(prefix="/users", tags=["users"])
异步 def get_service(db: AsyncSession = Depends(get_db)) -> UserService:
返回 UserService(db)
@router.get("", response_model=list[UserResponse])
异步 def list_users(
skip: int = 0,
limit: int = 100,
服务: UserService = Depends(get_service),
):
返回等待 服务.list(skip=skip, limit=limit)
@router.get("/{user_id}", response_model=UserResponse)
异步 def get_user(
user_id: int,
服务: UserService = Depends(get_service),
):
用户 = 等待 服务.get(user_id)
如果不 用户:
引发 HTTPException状态_code=404, detail="User not found"
返回 用户
@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
异步 def create_user(
user_in: UserCreate,
服务: UserService = Depends(get_service),
):
如果 等待 服务.get_by_email(user_in.email):
引发 HTTPException状态_code=400, detail="Email already registered"
返回等待 服务.create(user_in)
@router.patch("/{user_id}", response_model=UserResponse)
异步 def update_user(
user_id: int,
user_in: UserUpdate,
服务: UserService = Depends(get_service),
):
用户 = 等待 服务.get(user_id)
如果不 用户:
引发 HTTPException状态_code=404, detail="User not found"
返回等待 服务.update(user, user_in)
功能：users/init.py（导出）
从 app.features.users.api import router
从 app.features.users.models import User
从 app.features.users.schemas import UserCreate, UserResponse, UserUpdate
从 app.features.users.services import UserService
__all__ = ["router", "User", "UserCreate", "UserResponse", "UserUpdate", "UserService"]
主路由器（src/app/api/router.py）
从 fastapi import APIRouter
从 app.features.auth import router 作为 auth_router
从 app.features.users import router 作为 users_router
从 app.features.items import router 作为 items_router
api_router = APIRouter()
api_router.include_router(auth_router)
api_router.include_router(users_router)
api_router.include_router(items_router)
FastAPI 应用（src/app/main.py）
从 contextlib import asynccontextmanager
从 collections.abc import AsyncIterator
从 fastapi import FastAPI
从 fastapi.middleware.cors import CORSMiddleware
从 app.api.router import api_router
从 app.config import 设置
从 app.database import 引擎
@asynccontextmanager
异步 def lifespan(app: FastAPI) -> AsyncIterator[None]:
# 启动
产量
# 关闭
等待 引擎.dispose()
应用 = FastAPI(
标题=设置.app_name,
openapi_url=f"{设置.api_v1_prefix}/openapi.json",
lifespan=lifespan,
)
应用.add_middleware(
CORSMiddleware,
允许_origins=["*"], # 为生产配置
允许_credentials=True,
允许_methods=["*"],
允许_headers=["*"],
)
应用.include_router(api_router, prefix=设置.api_v1_prefix)
@应用.get("/health")
异步 def health_check():
返回 {"status": "healthy"}
数据库迁移与 Alembic
# 初始化 Alembic
uv run alembic init alembic
# 更新 alembic/env.py 为异步
# 然后创建迁移
uv run alembic revision --autogenerate -m "Initial migration"
# 应用迁移
uv run alembic upgrade head
异步 Alembic env.py
导入 asyncio
从 logging.config import fileConfig
从 sqlalchemy import pool
从 sqlalchemy.ext.asyncio import async_engine_from_config
从 alembic import context
从 app.config import 设置
从 app.models.base import Base
从 app.models import 用户, 项目 # 导入所有模型
配置 = context.config
配置.set_main_option("sqlalchemy.url", 设置.database_url)
如果 config.config_file_name 是 None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
防御 run_migrations_offline() -> None:
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
目标_metadata=target_metadata,
字面绑定=True,
方言_opts={"paramstyle": "named"},
)
与 context.begin_transaction():
context.run_migrations()
防御 do_run_migrations(connection) -> None:
context.configure(connection=connection, 目标_metadata=target_metadata)
与 context.begin_transaction():
context.run_migrations()
异步 def run_async_migrations() -> None:
connectable = async_engine_from_config(
config.get_section(config.config_ini_section, {}),
前缀="sqlalchemy.",
poolclass=pool.NullPool,
)
异步与 connectable.connect() 作为连接:
等待连接.run_sync(do_run_migrations)
等待 connectable.dispose()
防御 run_migrations_online() -> None:
asyncio.run(run_async_migrations())
如果 context.is_offline_mode():
run_migrations_offline()
否则:
run_migrations_online()
测试
conftest.py
导入 pytest
从 httpx import ASGITransport, AsyncClient
从 sqlalchemy.ext.asyncio import 创建_async_engine, async_sessionmaker, AsyncSession
从 app.main import 应用
从 app.database import get_db
从 app.models.base import Base
@pytest.fixture
异步 def db_session():
引擎 = 创建_async_engine(
"sqlite+aiosqlite:///:memory:",
echo=True,
)
异步与引擎.begin() 作为 conn:
等待 conn.run_sync(Base.metadata.create_all)
会话制造商 = async_sessionmaker(引擎, expire_on_commit=False)
异步与会话制造商() 作为会话:
产量会话
等待引擎.dispose()
@pytest.fixture
异步 def client(db_session: AsyncSession):
异步 def override_get_db():
产量 db_session
应用.dependency_overrides[get_db] = override_get_db
异步与 AsyncClient(
transport=ASGITransport(应用=应用),
base_url="http://test"
) 作为 ac:
产量 ac
应用.dependency_overrides.clear()
示例测试
导入 pytest
从 httpx import AsyncClient
@pytest.mark.asyncio
异步 def test_create_user(client: AsyncClient):
响应 = 等待 client.post(
"/api/v1/users",
json={
"email": "test@example.com",
"password": "testpassword123",
"full_name": "Test User",
},
)
断言响应.status_code == 201
数据 = 响应.json()
断言数据["email"] == "test@example.com"
断言 "id" 在 数据
运行应用程序
# 开发
uv run uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
# 生产
uv run uvicorn app.main:app --host 0.0.0.0 --port 8000 --works 4
# 运行测试
uv run pytest -v
# 运行覆盖
uv run pytest --cov=app --cov-report=html
# 代码检查和格式化
uv run ruff check .
uv run ruff format .
# 类型检查
uv run mypy src/
最佳实践
-
分层架构
- 路由：处理 HTTP，验证，响应格式化
- 服务：业务逻辑，编排
- 仓库：数据访问，查询
-
依赖注入
- 使用 FastAPI 的
Depends()进行清晰的 DI - 注入数据库会话，服务，配置
- 使用 FastAPI 的
-
类型安全
- 使用 Pydantic 用于所有请求/响应模式
- 使用 SQLAlchemy 2.0 映射列与类型
- 启用严格的 mypy
-
异步优先
- 整个过程中使用 async/await
- 使用 asyncpg 用于 PostgreSQL
- 使用 aiosqlite 用于测试
-
配置
- 使用 pydantic-settings 用于类型安全的配置
- 从环境变量中加载
- 永远不要提交秘密
-
测试
- 使用内存 SQLite 用于单元测试
- 使用测试容器用于集成测试
- 模拟外部服务