FastAPI 从入门到实践构建高性能 Python Web API 的完整指南一、为什么选择 FastAPI在Python的Web框架生态中我们已经有了Django、Flask等成熟方案。FastAPI的出现解决了什么痛点答案在于现代Web开发对性能、类型安全和开发效率的更高要求。性能优势基于Starlette用于Web处理和Pydantic用于数据验证FastAPI的性能表现接近Node.js和Go的框架。在I/O密集型的API服务场景中这种优势尤为明显。开发体验通过Python类型提示你可以在编码阶段就发现很多潜在错误而不是等到运行时。编辑器的自动补全变得异常精准这大大减少了调试时间。自动交互文档不需要额外编写文档FastAPI会自动生成符合OpenAPI标准的交互式文档。前端开发者可以直接在浏览器中测试接口这在团队协作中价值巨大。异步支持原生支持async/await让你能够轻松处理高并发请求而不需要引入复杂的多线程编程。但FastAPI并非万能钥匙。如果你的项目主要是传统的服务端渲染SSR或者需要大量的内置管理后台功能Django可能仍是更好的选择。FastAPI的核心优势在于构建API服务特别是微服务架构中的单个服务。## 二、环境准备与项目初始化bashmkdir fastapi-project cd fastapi-projectpython -m venv venvsource venv/bin/activatepip install fastapi uvicorn[standard]pip install sqlalchemy asyncpg alembicpip install pydantic pydantic-settingspip install redis httpx pytest### 2.1 项目结构fastapi-project/├── app/│ ├── __init__.py│ ├── main.py│ ├── config.py│ ├── database.py│ ├── models/│ │ └── user.py│ ├── schemas/│ │ └── user.py│ ├── api/│ │ └── v1/│ │ └── users.py│ ├── services/│ │ └── user_service.py│ └── middleware/│ └── auth.py├── tests/├── alembic/└── docker-compose.yml## 三、核心概念实战### 3.1 应用入口python# app/main.pyfrom fastapi import FastAPIfrom fastapi.middleware.cors import CORSMiddlewarefrom app.api.v1 import usersapp FastAPI( title电商平台API, description高性能电商平台后端服务, version1.0.0, docs_url/api/docs, redoc_url/api/redoc,)app.add_middleware( CORSMiddleware, allow_origins[http://localhost:3000], allow_credentialsTrue, allow_methods[*], allow_headers[*],)app.include_router(users.router, prefix/api/v1/users, tags[用户管理])app.get(/health)async def health_check(): return {status: healthy, version: 1.0.0}### 3.2 数据验证Pydantic Schemapython# app/schemas/user.pyfrom pydantic import BaseModel, Field, EmailStr, field_validatorfrom typing import Optionalfrom datetime import datetimeimport reclass UserCreate(BaseModel): username: str Field(..., min_length3, max_length50, description用户名) email: EmailStr Field(..., description邮箱地址) password: str Field(..., min_length8, max_length128, description密码) full_name: Optional[str] Field(None, max_length100) field_validator(username) classmethod def validate_username(cls, v: str) - str: if not re.match(r^[a-zA-Z0-9_]$, v): raise ValueError(用户名只能包含字母、数字和下划线) return v.lower() field_validator(password) classmethod def validate_password_strength(cls, v: str) - str: if not re.search(r[A-Z], v): raise ValueError(密码必须包含至少一个大写字母) if not re.search(r[a-z], v): raise ValueError(密码必须包含至少一个小写字母) if not re.search(r\d, v): raise ValueError(密码必须包含至少一个数字) if not re.search(r[!#$%^*(),.?:|], v): raise ValueError(密码必须包含至少一个特殊字符) return vclass UserResponse(BaseModel): id: int username: str email: str full_name: Optional[str] is_active: bool created_at: datetime model_config {from_attributes: True}class UserUpdate(BaseModel): full_name: Optional[str] Field(None, max_length100) email: Optional[EmailStr] None### 3.3 数据库模型SQLAlchemy 2.0python# app/database.pyfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmakerfrom sqlalchemy.orm import DeclarativeBasefrom app.config import settingsengine create_async_engine( settings.DATABASE_URL, echosettings.DEBUG, pool_size20, max_overflow10, pool_pre_pingTrue,)AsyncSessionLocal async_sessionmaker(engine, class_AsyncSession, expire_on_commitFalse)class Base(DeclarativeBase): passasync def get_db() - AsyncSession: async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception: await session.rollback() raise finally: await session.close()# app/models/user.pyfrom sqlalchemy import String, Boolean, DateTime, funcfrom sqlalchemy.orm import Mapped, mapped_columnfrom datetime import datetimefrom app.database import Baseclass User(Base): __tablename__ users id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) username: Mapped[str] mapped_column(String(50), uniqueTrue, indexTrue, nullableFalse) email: Mapped[str] mapped_column(String(255), uniqueTrue, indexTrue, nullableFalse) hashed_password: Mapped[str] mapped_column(String(255), nullableFalse) full_name: Mapped[str | None] mapped_column(String(100)) is_active: Mapped[bool] mapped_column(Boolean, defaultTrue) is_superuser: Mapped[bool] mapped_column(Boolean, defaultFalse) created_at: Mapped[datetime] mapped_column(DateTime(timezoneTrue), server_defaultfunc.now()) updated_at: Mapped[datetime] mapped_column(DateTime(timezoneTrue), server_defaultfunc.now(), onupdatefunc.now())### 3.4 API路由实现python# app/api/v1/users.pyfrom fastapi import APIRouter, Depends, HTTPException, Query, statusfrom sqlalchemy.ext.asyncio import AsyncSessionfrom app.database import get_dbfrom app.schemas.user import UserCreate, UserResponse, UserUpdatefrom app.services.user_service import UserServicerouter APIRouter()router.post(/, response_modelUserResponse, status_codestatus.HTTP_201_CREATED)async def create_user(user_data: UserCreate, db: AsyncSession Depends(get_db)): service UserService(db) existing await service.get_by_username(user_data.username) if existing: raise HTTPException(status_codestatus.HTTP_409_CONFLICT, detail用户名已存在) existing await service.get_by_email(user_data.email) if existing: raise HTTPException(status_codestatus.HTTP_409_CONFLICT, detail邮箱已被注册) return await service.create_user(user_data)router.get(/, response_modeldict)async def list_users( page: int Query(1, ge1), size: int Query(20, ge1, le100), search: str Query(None), is_active: bool Query(None), db: AsyncSession Depends(get_db),): service UserService(db) users, total await service.list_users(pagepage, sizesize, searchsearch, is_activeis_active) return { data: users, pagination: {page: page, size: size, total: total, total_pages: (total size - 1) // size} }router.get(/{user_id}, response_modelUserResponse)async def get_user(user_id: int, db: AsyncSession Depends(get_db)): service UserService(db) user await service.get_by_id(user_id) if not user: raise HTTPException(status_codestatus.HTTP_404_NOT_FOUND, detail用户不存在) return userrouter.patch(/{user_id}, response_modelUserResponse)async def update_user(user_id: int, user_data: UserUpdate, db: AsyncSession Depends(get_db)): service UserService(db) user await service.update_user(user_id, user_data) if not user: raise HTTPException(status_codestatus.HTTP_404_NOT_FOUND, detail用户不存在) return userrouter.delete(/{user_id}, status_codestatus.HTTP_204_NO_CONTENT)async def delete_user(user_id: int, db: AsyncSession Depends(get_db)): service UserService(db) success await service.soft_delete(user_id) if not success: raise HTTPException(status_codestatus.HTTP_404_NOT_FOUND, detail用户不存在)### 3.5 业务逻辑层python# app/services/user_service.pyfrom sqlalchemy.ext.asyncio import AsyncSessionfrom sqlalchemy import select, func, or_from passlib.context import CryptContextfrom app.models.user import Userfrom app.schemas.user import UserCreate, UserUpdatepwd_context CryptContext(schemes[bcrypt], deprecatedauto)class UserService: def __init__(self, db: AsyncSession): self.db db async def create_user(self, user_data: UserCreate) - User: user User( usernameuser_data.username, emailuser_data.email, hashed_passwordpwd_context.hash(user_data.password), full_nameuser_data.full_name, ) self.db.add(user) await self.db.flush() await self.db.refresh(user) return user async def get_by_id(self, user_id: int) - User | None: result await self.db.execute(select(User).where(User.id user_id, User.is_active True)) return result.scalar_one_or_none() async def get_by_username(self, username: str) - User | None: result await self.db.execute(select(User).where(User.username username)) return result.scalar_one_or_none() async def get_by_email(self, email: str) - User | None: result await self.db.execute(select(User).where(User.email email)) return result.scalar_one_or_none() async def list_users(self, page: int, size: int, search: str None, is_active: bool None): query select(User) count_query select(func.count(User.id)) if search: search_filter or_(User.username.ilike(f%{search}%), User.email.ilike(f%{search}%), User.full_name.ilike(f%{search}%)) query query.where(search_filter) count_query count_query.where(search_filter) if is_active is not None: query query.where(User.is_active is_active) count_query count_query.where(User.is_active is_active) total_result await self.db.execute(count_query) total total_result.scalar() query query.offset((page - 1) * size).limit(size).order_by(User.created_at.desc()) result await self.db.execute(query) return result.scalars().all(), total async def update_user(self, user_id: int, user_data: UserUpdate) - User | None: user await self.get_by_id(user_id) if not user: return None update_data user_data.model_dump(exclude_unsetTrue) for field, value in update_data.items(): setattr(user, field, value) await self.db.flush() await self.db.refresh(user) return user async def soft_delete(self, user_id: int) - bool: user await self.get_by_id(user_id) if not user: return False user.is_active False await self.db.flush() return True## 四、认证与授权python# app/middleware/auth.pyfrom fastapi import Depends, HTTPException, statusfrom fastapi.security import HTTPBearer, HTTPAuthorizationCredentialsfrom jose import JWTError, jwtfrom app.config import settingsfrom app.database import get_dbfrom app.services.user_service import UserServicesecurity HTTPBearer()async def get_current_user(credentials: HTTPAuthorizationCredentials Depends(security), db Depends(get_db)): token credentials.credentials credentials_exception HTTPException(status_codestatus.HTTP_401_UNAUTHORIZED, detail无法验证凭据, headers{WWW-Authenticate: Bearer}) try: payload jwt.decode(token, settings.SECRET_KEY, algorithms[settings.ALGORITHM]) user_id: int payload.get(sub) if user_id is None: raise credentials_exception except JWTError: raise credentials_exception service UserService(db) user await service.get_by_id(user_id) if user is None: raise credentials_exception return userasync def get_current_active_superuser(current_user Depends(get_current_user)): if not current_user.is_superuser: raise HTTPException(status_codestatus.HTTP_403_FORBIDDEN, detail权限不足) return current_user## 五、请求日志中间件python# app/middleware/logging.pyimport timeimport loggingfrom fastapi import Requestlogger logging.getLogger(__name__)async def log_requests(request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time logger.info(f{request.method} {request.url.path} 状态: {response.status_code} 耗时: {process_time:.3f}s) response.headers[X-Process-Time] str(process_time) return response## 六、测试策略python# tests/test_users.pyimport pytestfrom httpx import AsyncClient, ASGITransportfrom app.main import apppytest.mark.asyncioasync def test_create_user(): async with AsyncClient(transportASGITransport(appapp), base_urlhttp://test) as client: response await client.post(/api/v1/users/, json{ username: testuser, email: testexample.com, password: Test1234, full_name: Test User }) assert response.status_code 201 data response.json() assert data[username] testuser assert data[email] testexample.compytest.mark.asyncioasync def test_create_user_duplicate(): async with AsyncClient(transportASGITransport(appapp), base_urlhttp://test) as client: response await client.post(/api/v1/users/, json{ username: testuser, email: anotherexample.com, password: Test1234 }) assert response.status_code 409pytest.mark.asyncioasync def test_get_user_not_found(): async with AsyncClient(transportASGITransport(appapp), base_urlhttp://test) as client: response await client.get(/api/v1/users/99999) assert response.status_code 404pytest.mark.asyncioasync def test_list_users_pagination(): async with AsyncClient(transportASGITransport(appapp), base_urlhttp://test) as client: response await client.get(/api/v1/users/?page1size10) assert response.status_code 200 data response.json() assert data in data assert pagination in data assert data[pagination][page] 1 assert data[pagination][size] 10## 七、Docker部署dockerfile# DockerfileFROM python:3.12-slimWORKDIR /appCOPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txtCOPY . .EXPOSE 8000CMD [“uvicorn”, “app.main:app”, “–host”, “0.0.0.0”, “–port”, “8000”]yaml# docker-compose.ymlversion: 3.8’services: api: build: . ports: - “8000:8000” environment: - DATABASE_URLpostgresqlasyncpg://user:passworddb:5432/appdb - SECRET_KEYyour-secret-key depends_on: db: condition: service_healthy restart: unless-stopped db: image: postgres:16-alpine environment: POSTGRES_USER: user