FastAPI 实战：构建一个完整的 API 应用 – wiki基地

作者： /

FastAPI 实战：构建一个完整的 API 应用

FastAPI 凭借其高性能、易用性、自动生成文档等优势，已成为 Python Web API 开发的热门框架。本文将带你从零开始，使用 FastAPI 构建一个完整的 API 应用，涵盖项目初始化、数据模型定义、数据库集成、API 路由设计、认证授权、测试以及部署等关键环节，帮助你快速掌握 FastAPI 的实战技能。

一、项目初始化与环境配置

  1. 创建项目目录:
    bash
    mkdir fastapi_project
    cd fastapi_project

  2. 创建虚拟环境: 推荐使用 venv 或 conda 创建隔离的虚拟环境，避免依赖冲突。
    bash
    python3 -m venv venv
    source venv/bin/activate # Linux/macOS
    # venv\Scripts\activate # Windows

  3. 安装 FastAPI 及其依赖:
    bash
    pip install fastapi uvicorn python-multipart SQLAlchemy alembic

    • fastapi: FastAPI 框架本身。
    • uvicorn: ASGI (Asynchronous Server Gateway Interface) 服务器，用于运行 FastAPI 应用。
    • python-multipart: 用于处理文件上传，在需要处理 multipart/form-data 请求时必须安装。
    • SQLAlchemy: Python SQL 工具包和对象关系映射 (ORM) 库，用于数据库交互。
    • alembic: 数据库迁移工具，用于管理数据库结构变更。

  4. 项目目录结构:
    fastapi_project/
    ├── app/
    │ ├── __init__.py
    │ ├── database.py # 数据库连接和 ORM 模型定义
    │ ├── models.py # 数据库模型定义
    │ ├── schemas.py # Pydantic 数据模型定义
    │ ├── api/ # API 路由
    │ │ ├── __init__.py
    │ │ ├── items.py # Item 相关的 API
    │ │ ├── users.py # User 相关的 API
    │ ├── main.py # FastAPI 应用入口
    ├── alembic.ini # Alembic 配置文件
    ├── migrations/ # Alembic 迁移脚本
    ├── tests/ # 测试用例
    ├── README.md # 项目说明
    └── requirements.txt # 依赖列表

二、数据库集成与 ORM 模型定义

  1. 配置数据库连接: 在 app/database.py 中配置数据库连接信息。 这里以 SQLite 为例，方便快速上手。 在生产环境中，建议使用 PostgreSQL, MySQL 等更可靠的数据库。

    “`python

    app/database.py

    from sqlalchemy import create_engine
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker

    DATABASE_URL = “sqlite:///./test.db” # SQLite 数据库文件路径

    engine = create_engine(DATABASE_URL, connect_args={“check_same_thread”: False}) # SQLite 特定配置

    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    Base = declarative_base()

    Dependency to get the database session

    def get_db():
    db = SessionLocal()
    try:
    yield db
    finally:
    db.close()
    ``
    2. **定义 ORM 模型:** 在    app/models.py` 中定义数据库表结构对应的 ORM 模型。

    “`python

    app/models.py

    from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
    from sqlalchemy.orm import relationship
    from sqlalchemy.sql import func # for timestamp
    from .database import Base # 导入 Base

    class User(Base):
    tablename = “users”

    id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True, nullable=False)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
items = relationship("Item", back_populates="owner") # Relationship to items

    class Item(Base):
    tablename = “items”

    id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, nullable=True)
owner_id = Column(Integer, ForeignKey("users.id"))
created_at = Column(DateTime(timezone=True), server_default=func.now())
owner = relationship("User", back_populates="items") # Relationship to owner

    “`

  2. 使用 Alembic 进行数据库迁移:

    • 初始化 Alembic:
      bash
      alembic init migrations

    • 修改 alembic.ini:

      • 设置 sqlalchemy.url 指向你的数据库 URL。
      • 设置 script_location 指向 migrations 目录。

    • 生成初始迁移脚本:
      bash
      alembic revision --autogenerate -m "Create tables"

    • 应用迁移:
      bash
      alembic upgrade head

三、 Pydantic 数据模型定义

Pydantic 用于定义请求和响应的数据结构，提供数据验证和序列化/反序列化功能。 在app/schemas.py中定义：

“`python

app/schemas.py

from pydantic import BaseModel
from datetime import datetime

class UserBase(BaseModel):
username: str
email: str

class UserCreate(UserBase):
password: str

class User(UserBase):
id: int
is_active: bool
created_at: datetime

class Config:
    orm_mode = True

class ItemBase(BaseModel):
title: str
description: str | None = None #description can be null

class ItemCreate(ItemBase):
pass

class Item(ItemBase):
id: int
owner_id: int
created_at: datetime

class Config:
    orm_mode = True

“`

四、API 路由设计与实现

在 app/api/ 目录下创建不同的模块，用于组织不同资源的 API 路由。 例如：

  1. 用户 API (app/api/users.py):

    “`python

    app/api/users.py

    from fastapi import APIRouter, Depends, HTTPException
    from sqlalchemy.orm import Session
    from .. import models, schemas, database
    from .. import utils # 假设有一个 utils.py 用于处理密码哈希等

    router = APIRouter(prefix=”/users”, tags=[“users”])

    @router.post(“/”, response_model=schemas.User, status_code=201)
    def create_user(user: schemas.UserCreate, db: Session = Depends(database.get_db)):
    db_user = db.query(models.User).filter(models.User.email == user.email).first()
    if db_user:
    raise HTTPException(status_code=400, detail=”Email already registered”)

    hashed_password = utils.hash_password(user.password)
db_user = models.User(username=user.username, email=user.email, hashed_password=hashed_password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user

    @router.get(“/{user_id}”, response_model=schemas.User)
    def read_user(user_id: int, db: Session = Depends(database.get_db)):
    db_user = db.query(models.User).filter(models.User.id == user_id).first()
    if db_user is None:
    raise HTTPException(status_code=404, detail=”User not found”)
    return db_user

    @router.get(“/”, response_model=list[schemas.User])
    def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(database.get_db)):
    users = db.query(models.User).offset(skip).limit(limit).all()
    return users
    “`

  2. Item API (app/api/items.py):

    “`python

    app/api/items.py

    from fastapi import APIRouter, Depends, HTTPException
    from sqlalchemy.orm import Session
    from .. import models, schemas, database

    router = APIRouter(prefix=”/items”, tags=[“items”])

    @router.post(“/”, response_model=schemas.Item, status_code=201)
    def create_item(item: schemas.ItemCreate, db: Session = Depends(database.get_db), current_user: schemas.User = Depends(utils.get_current_user)): # 使用 Depends 获取当前用户
    db_item = models.Item(**item.dict(), owner_id=current_user.id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

    @router.get(“/{item_id}”, response_model=schemas.Item)
    def read_item(item_id: int, db: Session = Depends(database.get_db)):
    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if db_item is None:
    raise HTTPException(status_code=404, detail=”Item not found”)
    return db_item

    @router.get(“/”, response_model=list[schemas.Item])
    def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(database.get_db)):
    items = db.query(models.Item).offset(skip).limit(limit).all()
    return items
    “`

  3. 认证授权 （app/utils.py）

“`python

app/utils.py

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from datetime import datetime, timedelta
from passlib.context import CryptContext
from sqlalchemy.orm import Session
from . import schemas, models, database
from .config import settings #假设你有一个config.py来存放一些配置信息

pwd_context = CryptContext(schemes=[“bcrypt”], deprecated=”auto”)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl=”login”) #定义tokenUrl

密码hash

def hash_password(password: str):
return pwd_context.hash(password)

验证密码

def verify_password(plain_password: str, hashed_password: str):
return pwd_context.verify(plain_password, hashed_password)

创建access token

def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({“exp”: expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt

获取当前用户

async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(database.get_db)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=”Could not validate credentials”,
headers={“WWW-Authenticate”: “Bearer”},
)
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
user_id: str = payload.get(“user_id”)
if user_id is None:
raise credentials_exception
token_data = schemas.TokenData(user_id=user_id) #创建一个TokenData的Schema
except JWTError:
raise credentials_exception
user = db.query(models.User).filter(models.User.id == token_data.user_id).first()
if user is None:
raise credentials_exception
return user

“`

  1. 在 app/main.py 中注册路由:

    “`python

    app/main.py

    from fastapi import FastAPI
    from .api import users, items
    from .database import engine
    from . import models

    models.Base.metadata.create_all(bind=engine) # 创建数据库表 (生产环境建议使用 Alembic)

    app = FastAPI()

    app.include_router(users.router)
    app.include_router(items.router)
    “`

五、运行应用

bash
uvicorn app.main:app --reload
访问 http://127.0.0.1:8000/docs 查看自动生成的 API 文档。

六、测试

使用 pytest 或其他测试框架编写测试用例，确保 API 的功能正确性和稳定性。 可以针对不同的 API 端点，编写单元测试和集成测试。

七、部署

将 FastAPI 应用部署到生产环境，例如使用 Docker 容器化部署到云服务器。常用的部署方案包括：

  • Docker + Docker Compose: 使用 Docker 容器化应用，并使用 Docker Compose 管理多个容器。
  • Kubernetes: 使用 Kubernetes 进行容器编排和管理。
  • 云平台服务: 利用 AWS、Azure、Google Cloud 等云平台提供的 PaaS 服务进行部署，例如 AWS Elastic Beanstalk, Azure App Service, Google App Engine。

八、总结

本文详细介绍了使用 FastAPI 构建完整 API 应用的步骤，涵盖了项目初始化、数据模型定义、数据库集成、API 路由设计、认证授权、测试以及部署等关键环节。通过实践，你可以掌握 FastAPI 的核心概念和使用方法，并能够构建高性能、易维护的 API 应用。

九、一些补充说明和最佳实践

  1. 异常处理: 使用 FastAPI 的异常处理机制，统一处理 API 异常，返回友好的错误信息。
  2. 日志记录: 配置日志记录，方便调试和监控 API 应用。
  3. API 版本控制: 使用 URL 前缀或请求头进行 API 版本控制，方便 API 的升级和维护。
  4. 缓存: 使用 Redis 或 Memcached 等缓存服务，提升 API 的响应速度。
  5. API 文档: 充分利用 FastAPI 自动生成的 API 文档，方便开发者使用你的 API。
  6. 安全性: 关注 API 的安全性，例如使用 HTTPS 加密传输，防止 CSRF 攻击等。 使用JWT Token进行身份验证。
  7. 数据库连接池: 在生产环境中，使用数据库连接池，提高数据库的并发处理能力。
  8. 配置文件管理: 使用配置文件管理工具，例如 python-decouple，方便管理不同环境下的配置信息。
  9. 异步编程: 充分利用 FastAPI 的异步特性，处理 IO 密集型任务，提高 API 的性能。

希望这篇文章能够帮助你更好地理解和使用 FastAPI，构建出优秀的 API 应用。 记住，实践是最好的老师！

发表评论

您的邮箱地址不会被公开。 必填项已用 * 标注

滚动至顶部