Skip to main content

FastAPI 面试

以下是一些与 FastAPI 相关的常见面试问题,涵盖基础概念、实践经验和高级特性,帮助应聘者准备技术面试:


1. 基础概念

  1. 什么是 FastAPI?它的主要优点是什么?

    • FastAPI 是一个基于 Python 的现代、高性能 Web 框架,用于构建 API。
    • 优点:
      • 高性能(基于 Starlette 和 Pydantic)。
      • 自动生成交互式 API 文档(Swagger UI 和 ReDoc)。
      • 基于 Python 类型提示(Type Hints)的自动数据验证和序列化。
      • 支持异步编程(Async/Await)。
  2. FastAPI 和 Flask/Django 的区别?

    • FastAPI 天生支持异步,性能更高;Flask/Django(同步为主,需插件支持异步)。
    • FastAPI 内置数据验证和文档生成;Flask 需要手动扩展(如 Marshmallow)。
    • Django 是“全栈”框架(自带 ORM、Admin 等),FastAPI 更轻量,专注 API。
  3. 什么是 Pydantic?它在 FastAPI 中的作用?

    • Pydantic 是一个基于类型提示的数据验证库,FastAPI 用它来处理请求和响应的数据模型(如请求体、查询参数)。
    • 示例:定义 User 模型时自动验证字段类型和约束。

2. 路由与请求处理

  1. 如何定义路由(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}
  2. 路径参数、查询参数、请求体的区别?如何定义它们?

    • 路径参数/items/{item_id}(必需参数)。
    • 查询参数/items/?skip=0&limit=10(可选,通过函数参数声明)。
    • 请求体:POST/PUT 请求中传递的 JSON 数据(通过 Pydantic 模型接收)。
  3. 如何处理文件上传?

    from fastapi import UploadFile, File

    @app.post("/upload/")
    async def upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

3. 数据验证与序列化

  1. 如何用 Pydantic 定义请求/响应模型?

    from pydantic import BaseModel

    class Item(BaseModel):
    name: str
    price: float

    @app.post("/items/")
    async def create_item(item: Item):
    return item # 自动序列化为 JSON
  2. 如何添加自定义验证(如字段长度、正则表达式)?
    使用 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)

  1. 什么是依赖注入?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()
  2. 依赖注入的常见应用场景?

    • 数据库会话管理。
    • 用户认证(如 JWT 验证)。
    • 复用公共逻辑(如分页参数)。

5. 异步支持

  1. FastAPI 如何支持异步?什么情况下应该用 async def

    • FastAPI 基于 Starlette,原生支持 async/await
    • 在 I/O 密集型操作(如数据库查询、外部 API 调用)时使用异步,阻塞操作(CPU 密集型)可能仍需多进程。
  2. 如何调用同步函数(如传统数据库库 SQLAlchemy)?

    • 使用 asyncio.run_in_executor 在线程池中运行同步代码,避免阻塞事件循环。

6. 中间件与异常处理

  1. 如何添加中间件(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
  2. 如何自定义异常处理器?

    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. 安全与认证

  1. 如何实现 JWT 认证?
    使用 OAuth2PasswordBearerpython-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"}
  2. 如何防止 SQL 注入?

    • 使用 ORM(如 SQLAlchemy)的参数化查询,避免直接拼接 SQL 字符串。

8. 测试与部署

  1. 如何测试 FastAPI 应用?

    • 使用 TestClient(来自 fastapi.testclient):
    from fastapi.testclient import TestClient

    client = TestClient(app)
    response = client.get("/items/1")
    assert response.status_code == 200
  2. 如何部署 FastAPI?常见工具有哪些?

    • ASGI 服务器:Uvicorn、Hypercorn。
    • 容器化:Docker + Gunicorn(多进程)。
    • 云服务:AWS Lambda(通过 Mangum)、Kubernetes。

9. 高级问题

  1. 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"}
  2. 如何优化 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