FastAPI 面试
以下是一些与 FastAPI 相关的常见面试问题,涵盖基础概念、实践经验和高级特性,帮助应聘者准备技术面试:
1. 基础概念
-
什么是 FastAPI?它的主要优点是什么?
- FastAPI 是一个基于 Python 的现代、高性能 Web 框架,用于构建 API。
- 优点:
- 高性能(基于 Starlette 和 Pydantic)。
- 自动生成交互式 API 文档(Swagger UI 和 ReDoc)。
- 基于 Python 类型提示(Type Hints)的自动数据验证和序列化。
- 支持异步编程(Async/Await)。
-
FastAPI 和 Flask/Django 的区别?
- FastAPI 天生支持异步,性能更高;Flask/Django(同步为主,需插件支持异步)。
- FastAPI 内置数据验证和文档生成;Flask 需要手动扩展(如 Marshmallow)。
- Django 是“全栈”框架(自带 ORM、Admin 等),FastAPI 更轻量,专注 API。
-
什么是 Pydantic?它在 FastAPI 中的作用?
- Pydantic 是一个基于类型提示的数据验证库,FastAPI 用它来处理请求和响应的数据模型(如请求体、查询参数)。
- 示例:定义
User
模型时自动验证字段类型和约束。
2. 路由与请求处理
-
如何定义路由(Route)和 HTTP 方法(GET/POST)?
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id} -
路径参数、查询参数、请求体的区别?如何定义它们?
- 路径参数:
/items/{item_id}
(必需参数)。 - 查询参数:
/items/?skip=0&limit=10
(可选,通过函数参数声明)。 - 请求体:POST/PUT 请求中传递的 JSON 数据(通过 Pydantic 模型接收)。
- 路径参数:
-
如何处理文件上传?
from fastapi import UploadFile, File
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
return {"filename": file.filename}
3. 数据验证与序列化
-
如何用 Pydantic 定义请求/响应模型?
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
@app.post("/items/")
async def create_item(item: Item):
return item # 自动序列化为 JSON -
如何添加自定义验证(如字段长度、正则表达式)?
使用 Pydantic 的Field
或验证器:from pydantic import Field, validator
class User(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
@validator("username")
def validate_username(cls, v):
if "admin" in v:
raise ValueError("Username cannot contain 'admin'")
return v
4. 依赖注入(Dependency Injection)
-
什么是依赖注入?FastAPI 中如何实现?
-
通过
Depends
声明依赖(如数据库连接、权限验证)。 -
示例:
from fastapi import Depends
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/")
async def get_users(db: Session = Depends(get_db)):
return db.query(User).all()
-
-
依赖注入的常见应用场景?
- 数据库会话管理。
- 用户认证(如 JWT 验证)。
- 复用公共逻辑(如分页参数)。
5. 异步支持
-
FastAPI 如何支持异步?什么情况下应该用
async def
?- FastAPI 基于 Starlette,原生支持
async/await
。 - 在 I/O 密集型操作(如数据库查询、外部 API 调用)时使用异步,阻塞操作(CPU 密集型)可能仍需多进程。
- FastAPI 基于 Starlette,原生支持
-
如何调用同步函数(如传统数据库库 SQLAlchemy)?
- 使用
asyncio.run_in_executor
在线程池中运行同步代码,避免阻塞事件循环。
- 使用
6. 中间件与异常处理
-
如何添加中间件(Middleware)?
from fastapi import Request
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
response = await call_next(request)
response.headers["X-Custom-Header"] = "Example"
return response -
如何自定义异常处理器?
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
@app.exception_handler(HTTPException)
async def custom_exception_handler(request: Request, exc: HTTPException):
return JSONResponse({"error": exc.detail}, status_code=exc.status_code)
7. 安全与认证
-
如何实现 JWT 认证?
使用OAuth2PasswordBearer
和python-jose
库:from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.get("/protected/")
async def protected_route(token: str = Depends(oauth2_scheme)):
# 验证 token
return {"data": "secret"} -
如何防止 SQL 注入?
- 使用 ORM(如 SQLAlchemy)的参数化查询,避免直接拼接 SQL 字符串。
8. 测试与部署
-
如何测试 FastAPI 应用?
- 使用
TestClient
(来自fastapi.testclient
):
from fastapi.testclient import TestClient
client = TestClient(app)
response = client.get("/items/1")
assert response.status_code == 200 - 使用
-
如何部署 FastAPI?常见工具有哪些?
- ASGI 服务器:Uvicorn、Hypercorn。
- 容器化:Docker + Gunicorn(多进程)。
- 云服务:AWS Lambda(通过 Mangum)、Kubernetes。
9. 高级问题
-
FastAPI 如何实现后台任务(Background Tasks)?
from fastapi import BackgroundTasks
def log_request(message: str):
with open("log.txt", "a") as f:
f.write(message)
@app.post("/send-notification/")
async def send_notification(background_tasks: BackgroundTasks):
background_tasks.add_task(log_request, "Notification sent")
return {"message": "Processing in background"} -
如何优化 FastAPI 应用的性能?
- 启用异步数据库驱动(如
asyncpg
)。 - 使用缓存(如 Redis)。
- 减少中间件数量,避免阻塞操作。
- 启用异步数据库驱动(如
10. 开放性问题
- 你在项目中用 FastAPI 遇到过哪些挑战?如何解决的?
- 为什么选择 FastAPI 而不是其他框架?
- 如何设计一个遵循 RESTful 原则的 FastAPI 项目结构?
通过这些问题,面试官可以评估应聘者对 FastAPI 的理解深度和实际经验。建议结合具体项目经验回答,并准备代码示例。
11. 与 SQLALchemy, Alembic 和 Pydantic 整合
在 FastAPI 中使用 SQLAlchemy(ORM)、Alembic(数据库迁移)和 Pydantic(数据验证)是一个常见的企业级开发模式。以下是详细的整合方案,涵盖 项目结构设计、关键代码实现 和 最佳实践:
1. 项目结构推荐
project/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 主入口
│ ├── models.py # SQLAlchemy 模型
│ ├── schemas.py # Pydantic 模型
│ ├── database.py # 数据库连接和会话管理
│ ├── crud.py # 数据库操作(增删改查)
│ ├── dependencies.py # 依赖注入(如数据库会话)
│ └── migrations/ # Alembic 迁移脚本(自动生成)
├── alembic.ini # Alembic 配置文件
└── requirements.txt
2. 核心组件整合步骤
2.1 安装依赖
pip install fastapi sqlalchemy alembic pydantic psycopg2-binary # PostgreSQL驱动
2.2 配置 SQLAlchemy (database.py
)
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base() # ORM 模型基类
# 依赖注入:为每个请求生成独立会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
2.3 定义 ORM 模型 (models.py
)
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50))
email = Column(String(100), unique=True)
2.4 定义 Pydantic 模型 (schemas.py
)
from pydantic import BaseModel, EmailStr
class UserCreate(BaseModel): # 请求体验证
name: str
email: EmailStr
class UserResponse(BaseModel): # 响应模型
id: int
name: str
email: str
class Config:
from_attributes = True # 允许从ORM对象转换(旧版叫 orm_mode)
2.5 CRUD 操作 (crud.py
)
from sqlalchemy.orm import Session
from models import User
from schemas import UserCreate
def create_user(db: Session, user: UserCreate):
db_user = User(**user.model_dump()) # Pydantic -> ORM
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def get_user(db: Session, user_id: int):
return db.query(User).filter(User.id == user_id).first()
2.6 FastAPI 路由 (main.py
)
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from models import Base
from database import engine, get_db
from schemas import UserCreate, UserResponse
from crud import create_user, get_user
app = FastAPI()
# 创建数据库表(仅开发使用,生产用Alembic)
Base.metadata.create_all(bind=engine)
@app.post("/users/", response_model=UserResponse)
def create_user_endpoint(user: UserCreate, db: Session = Depends(get_db)):
return create_user(db, user)
@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = get_user(db, user_id)
if not db_user:
raise HTTPException(status_code=404, detail="User not found")
return db_user
3. 集成 Alembic 进行数据库迁移
3.1 初始化 Alembic
alembic init migrations
3.2 修改 alembic.ini
配置数据库URL
sqlalchemy.url = postgresql://user:password@localhost/dbname
3.3 修改 migrations/env.py
指向模型
from models import Base # 添加这行
target_metadata = Base.metadata # 修改这行
3.4 生成迁移脚本
alembic revision --autogenerate -m "create users table"
alembic upgrade head # 执行迁移
4. 关键整合技术点
4.1 SQLAlchemy 会话管理
- 使用
Depends(get_db)
为每个请求创建独立会话,自动关闭连接。 - 避免全局会话导致并发问题。
4.2 Pydantic 与 ORM 模型分离
- 请求/响应模型 (
schemas.py
):定义输入输出数据结构,严格校验。 - ORM 模型 (
models.py
):只关注数据库表结构,不混入业务逻辑。
4.3 Alembic 自动迁移
- 通过
--autogenerate
检测模型变更生成迁移脚本。 - 生产环境必须使用迁移脚本,禁止直接
Base.metadata.create_all()
。
4.4 响应模型序列化
response_model=UserResponse
确保输出数据经过 Pydantic