概述

API是连接前后端的前梁，当然前端既可以是UI web的前端，也可以是client端，API的测试和构建一样重要。
FastAPI 是一个用python构建API非常不错的框架，如何确保在客户端或服务器发生错误时 API 返回有用的响应？您应该针对真实数据进行测试还是模拟测试？
在本文中，介绍构建增删改查API(增加(Create)、检索(Retrieve)、更新(Update)和删除(Delete))使用FastAPI、SQLite并使用fixture等对pytest进行测试。

这里介绍构建Rest API - 从数据库创建、读取、更新、删除用户。

目标

结合本文，从以下几个方面介绍

1. 开发一个增删改查使用FastAPI框架在Python中实现Rest API
2. 使用SQLAIchemy ORM 工具和SQLite数据库交互
3. 使用pytest为FastAPI 进行单测
4. 处理错误和响应
5. 使用 FastAPI 的内置Swagger 记录REST API

对于API这里不在赘述，我们了解下FastAPI

FASTAPI 介绍

FastAPI是一个高性能的python web框架，可以轻松构建 API，最初由Sebastian Ramirez于18年创建，并在2019年发布。它是建立在Python 库之上：Starlette和Pydantic，其中Starlette 是提供底层 Web 应用程序框架，而 Pydantic 是用于数据验证和序列化的库。FastAPI的设计注重易用性，和性能，同时还内置了async/await的支持，使其比传统的同步线程模型更高效。

CRUD API

它是包括了HTTP方法（POST、GET、PUT、DELETE）的设计原则，用于对数据库DB系统中的数据维护的基本操作。广泛用于 Web 开发，用于实现内容管理和维护

项目设置

项目中，我们创建一个增删改查使用API从的关系型数据库（使用SQLite）创建、读取、更新和删除用户，项目名称这里叫fastapi_curdapi

fastapi_curdapi
├── app
│   ├── __init__.py
│   ├── database.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   └── user.py
├── pyest.ini
├── pyproject.toml
├── requirements.txt
├── tests
│   ├── __init__.py
│   ├── conftest.py
│   └── test_curd_api.py

其中app目录下包含源代码如下所示：

• database.py — 创建数据库引擎和 SQLite 设置。
• main.py — 创建 FastAPI 客户端、健康检查器和中间件。
• models.py - 数据库架构。
• schemas.py — 用户基础架构和响应架构。
• user.py — API 路由和响应格式。
• tests目录包含 API 的单元测试。

文件中列出了依赖项 pyproject.toml，或者说使用requirement.txt 这里使用pip3进行维护

fastapi==0.111.0
pydantic==2.7.3
SQLAlchemy==2.0.30
SQLAlchemy_Utils==0.41.2

对于项目生成requirement.txt，常用有两种方法

freeze

• 应用场景：在单一虚拟环境下，可以使用这种方式。
• 优点：这种方式会把当前环境下的所有依赖包都整理出来。
• 缺点：不是针对某个项目生成，如果当前环境下有多个项目，那么会生成大量无关的依赖信息。

pip freeze > requirements.txt

• 但是用这个方法，可以实现一个功能：删除当前环境的所有python依赖。

pip uninstall -r requirements.txt -y

pipreqs

• 应用场景：针对单个项目生成 requirements.txt
• 优点：使用 pipreqs 可以自动检索到当前项目下的所有组件及其版本，并生成 requirements.txt 文件，极大方便了项目迁移和部署的包管理。
• 缺点：相比直接用 freeze 命令，能直接隔离其它项目的包生成。

pipreqs ./ --encoding=utf-8
#强制执行命令 --force ，覆盖原有的 requirements.txt 文件
pipreqs ./ --encoding=utf-8 --force

所以这里使用pipreqps

代码介绍

database.py是创建数据库引擎和SQLite设置的代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This module is ***
# @Time : 2024/6/6 09:54
# @Author :
# function :
# @File : database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmakerSQLITE_DATABASE_URL = "sqlite:///./user.db"engine = create_engine(SQLITE_DATABASE_URL, echo=True, connect_args={"check_same_thread": False})SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)Base = declarative_base()def get_db():db = SessionLocal()try:yield dbfinally:db.close()

SQLite作为内存数据库，使用起来较为简便，当然还可以使用PostgreSQL、MySQL等数据库。而get_db函数是一个依赖项，将会为注入的每个请求创建一个新的数据库会话session。

models.py-数据库架构

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This module is ***
# @Time : 2024/6/6 09:54
# @Author : 
# function :
# @File : models.py
import uuidfrom sqlalchemy import Column, String, Boolean, TIMESTAMP
from sqlalchemy_utils import UUIDType
from sqlalchemy.sql import funcfrom app.database import Baseclass User(Base):__tablename__ ="users"# Primary key and GUID typeid = Column(UUIDType(binary=False), primary_key=True, default=uuid.uuid4())# String types with appropriate non0null constraintsfirst_name = Column(String(255), nullable=False, index=True)  # Index for faster searcheslast_name = Column(String(255), nullable=False, index=True)  # Index for faster searchesaddress = Column(String(255), nullable=True)# Boolean type with a deafult valueactivated = Column(Boolean, nullable=False, default=True)# Timestamp with timezone supportcreateAt = Column(TIMESTAMP(timezone=True), nullable=False, server_default=func.now())udpatedAt = Column(TIMESTAMP(timezone=True), default=None, onupdate=func.now())

本例中的数据模型非常简单 - 一个包含列的User表- id、first_name、last_name、address、activated、createdAt、updatedAt

使用SQLAlchemy ORM，我们定义表模式和列。

现在我们有了数据库模型，让我们使用Pydantic User创建和响应模型。
schemas.py — 用户基础架构和响应架构

from enum import Enum
from datetime import datetime
from typing import List
from pydantic import BaseModel, Field
from uuid import UUIDclass UserBaseSchema(BaseModel):id: UUID = Nonefirst_name: str = Field(..., description="The first name of the user", example="John")last_name: str = Field(..., description="The last name of the user", example="Doe")address: str = Noneactivated: bool = FalsecreatedAt: datetime = None# updatedAt: datetime = NoneupdatedAt: datetime = Field(default_factory=datetime.utcnow)class Config:from_attributes = Truepopulate_by_name = Truearbitrary_types_allowed = Trueclass Status(Enum):Success = "Success"Failed = "Failed"class UserResponse(BaseModel):Status: StatusUser: UserBaseSchemaclass GetUserResponse(BaseModel):Status: StatusUser: UserBaseSchemaclass ListUserResponse(BaseModel):status: Statusresults: intusers: List[UserBaseSchema]class DeleteUserResponse(BaseModel):Status: StatusMessage: str

上面的用法中使用到了 Pydantic 模型，用于在API路由中验证请求和响应负载信息

user.py- API路由和响应信息

import app.schemas as schemas
import app.models as models
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from fastapi import Depends, HTTPException, status, APIRouter
from app.database import get_dbrouter = APIRouter()@router.post("/", status_code=status.HTTP_201_CREATED, response_model=schemas.UserResponse
)
def create_user(payload: schemas.UserBaseSchema, db: Session = Depends(get_db)):try:# Create a new user instance from the payloadnew_user = models.User(**payload.model_dump())db.add(new_user)db.commit()db.refresh(new_user)except IntegrityError as e:db.rollback()# Log the error or handle it as neededraise HTTPException(status_code=status.HTTP_409_CONFLICT,detail="A user with the given details already exists.",) from eexcept Exception as e:db.rollback()# Handle other types of database errorsraise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,detail="An error occurred while creating the user.",) from e# Convert the SQLAlchemy model instance to a Pydantic modeluser_schema = schemas.UserBaseSchema.from_orm(new_user)# Return the successful creation responsereturn schemas.UserResponse(Status=schemas.Status.Success, User=user_schema)@router.get("/{userId}", status_code=status.HTTP_200_OK, response_model=schemas.GetUserResponse
)
def get_user(userId: str, db: Session = Depends(get_db)):user_query = db.query(models.User).filter(models.User.id == userId)db_user = user_query.first()if not db_user:raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f"No User with this id: `{userId}` found",)try:return schemas.GetUserResponse(Status=schemas.Status.Success, User=schemas.UserBaseSchema.model_validate(db_user))except Exception as e:raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,detail="An unexpected error occurred while fetching the user.",) from e@router.patch("/{userId}",status_code=status.HTTP_202_ACCEPTED,response_model=schemas.UserResponse,
)
def update_user(userId: str, payload: schemas.UserBaseSchema, db: Session = Depends(get_db)
):user_query = db.query(models.User).filter(models.User.id == userId)db_user = user_query.first()if not db_user:raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f"No User with this id: `{userId}` found",)try:update_data = payload.dict(exclude_unset=True)user_query.update(update_data, synchronize_session=False)db.commit()db.refresh(db_user)user_schema = schemas.UserBaseSchema.model_validate(db_user)return schemas.UserResponse(Status=schemas.Status.Success, User=user_schema)except IntegrityError as e:db.rollback()raise HTTPException(status_code=status.HTTP_409_CONFLICT,detail="A user with the given details already exists.",) from eexcept Exception as e:db.rollback()raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,detail="An error occurred while updating the user.",) from e@router.delete("/{userId}",status_code=status.HTTP_202_ACCEPTED,response_model=schemas.DeleteUserResponse,
)
def delete_user(userId: str, db: Session = Depends(get_db)):try:user_query = db.query(models.User).filter(models.User.id == userId)user = user_query.first()if not user:raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f"No User with this id: `{userId}` found",)user_query.delete(synchronize_session=False)db.commit()return schemas.DeleteUserResponse(Status=schemas.Status.Success, Message="User deleted successfully")except Exception as e:db.rollback()raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,detail="An error occurred while deleting the user.",) from e@router.get("/", status_code=status.HTTP_200_OK, response_model=schemas.ListUserResponse
)
def get_users(db: Session = Depends(get_db), limit: int = 10, page: int = 1, search: str = ""
):skip = (page - 1) * limitusers = (db.query(models.User).filter(models.User.first_name.contains(search)).limit(limit).offset(skip).all())return schemas.ListUserResponse(status=schemas.Status.Success, results=len(users), users=users)

上述代码定了 C R U D用户的API路由信息，可以更具需要再处理错误信息，日志记录，响应格式等方面的复杂功能。主要围绕4条信息开展的，也是我么日常使用较多的错误码类型。

• create_user- 创建新用户。201成功或409发生冲突时返回状态代码。
• get_user- 通过 ID 获取用户。如果200成功则返回状态代码，404如果未找到则返回状态代码。
• update_user- 通过 ID 更新用户。202成功或409发生冲突时返回状态代码。
• delete_user- 根据 ID 删除用户。如果202成功则返回状态代码，404如果未找到则返回状态代码。
• get_users- 获取用户列表。200成功时返回状态代码。

main.py — 创建 FastAPI 客户端、健康检查器和中间件

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This module is ***
# @Time : 2024/6/6 09:54
# @Author : 
# function :
# @File : main.py
from app import models, user
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.database import enginemodels.Base.metadata.create_all(bind=engine)app = FastAPI()origins = ["http://localhost:3000",
]app.add_middleware(CORSMiddleware,allow_origins=origins,allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)app.include_router(user.router, tags=["Users"], prefix="/api/users")@app.get("/api/healthchecker")
def root():return {"message": "The API is LIVE!!"}

上面的信息创建了FastAPI client，设置了 CORS中间件并定义了API路由user.py

run API

要拉起服务，执行如下cli命令

uvicorn app.main:app --host localhost --port 8000 --reload

在浏览器输入：http://localhost:8000/docs

通过使用不同接口完成测试。

pytest测试

tests使用conftest.py完成封装，使用test_curd_api完成业务测试。在conftest.py中

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This module is ***
# @Time : 2024/6/6 09:54
# @Author : 
# function :
# @File : conftest.py
import pytest
import uuid
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from fastapi.testclient import TestClient
from app.main import app
from app.database import Base, get_db# SQLite database URL for testing
SQLITE_DATABASE_URL = "sqlite:///./test_db.db"# Create a SQLAlchemy engine
engine = create_engine(SQLITE_DATABASE_URL,connect_args={"check_same_thread": False},poolclass=StaticPool,
)# Create a sessionmaker to manage sessions
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# Create tables in the database
Base.metadata.create_all(bind=engine)@pytest.fixture(scope="function")
def db_session():"""Create a new database session with a rollback at the end of the test."""connection = engine.connect()transaction = connection.begin()session = TestingSessionLocal(bind=connection)yield sessionsession.close()transaction.rollback()connection.close()@pytest.fixture(scope="function")
def test_client(db_session):"""Create a test client that uses the override_get_db fixture to return a session."""def override_get_db():try:yield db_sessionfinally:db_session.close()app.dependency_overrides[get_db] = override_get_dbwith TestClient(app) as test_client:yield test_client# Fixture to generate a random user id
@pytest.fixture()
def user_id() -> uuid.UUID:"""Generate a random user id."""return str(uuid.uuid4())# Fixture to generate a user payload
@pytest.fixture()
def user_payload(user_id):"""Generate a user payload."""return {"id": user_id,"first_name": "John","last_name": "Doe","address": "123 Farmville",}@pytest.fixture()
def user_payload_updated(user_id):"""Generate an updated user payload."""return {"first_name": "Jane","last_name": "Doe","address": "321 Farmville","activated": True,}

接着进行测试用例的编写

tests/test_crud_api.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This module is ***
# @Time : 2024/6/6 09:55
# @Author : 
# function :
# @File : test_curd_api.py
import timedef test_root(test_client):response = test_client.get("/api/healthchecker")assert response.status_code == 200assert response.json() == {"message": "The API is LIVE!!"}def test_create_get_user(test_client, user_payload):response = test_client.post("/api/users/", json=user_payload)response_json = response.json()assert response.status_code == 201# Get the created userresponse = test_client.get(f"/api/users/{user_payload['id']}")assert response.status_code == 200response_json = response.json()assert response_json["Status"] == "Success"assert response_json["User"]["id"] == user_payload["id"]assert response_json["User"]["address"] == "123 Farmville"assert response_json["User"]["first_name"] == "John"assert response_json["User"]["last_name"] == "Doe"def test_create_update_user(test_client, user_payload, user_payload_updated):response = test_client.post("/api/users/", json=user_payload)response_json = response.json()assert response.status_code == 201# Update the created usertime.sleep(1)  # Sleep for 1 second to ensure updatedAt is different (datetime precision is low in SQLite)response = test_client.patch(f"/api/users/{user_payload['id']}", json=user_payload_updated)response_json = response.json()assert response.status_code == 202assert response_json["Status"] == "Success"assert response_json["User"]["id"] == user_payload["id"]assert response_json["User"]["address"] == "321 Farmville"assert response_json["User"]["first_name"] == "Jane"assert response_json["User"]["last_name"] == "Doe"assert response_json["User"]["activated"] is Trueassert (response_json["User"]["updatedAt"] is not Noneand response_json["User"]["updatedAt"] > response_json["User"]["createdAt"])def test_create_delete_user(test_client, user_payload):response = test_client.post("/api/users/", json=user_payload)response_json = response.json()assert response.status_code == 201# Delete the created userresponse = test_client.delete(f"/api/users/{user_payload['id']}")response_json = response.json()assert response.status_code == 202assert response_json["Status"] == "Success"assert response_json["Message"] == "User deleted successfully"# Get the deleted userresponse = test_client.get(f"/api/users/{user_payload['id']}")assert response.status_code == 404response_json = response.json()assert response_json["detail"] == f"No User with this id: `{user_payload['id']}` found"def test_get_user_not_found(test_client, user_id):response = test_client.get(f"/api/users/{user_id}")assert response.status_code == 404response_json = response.json()assert response_json["detail"] == f"No User with this id: `{user_id}` found"def test_create_user_wrong_payload(test_client):response = test_client.post("/api/users/", json={})assert response.status_code == 422def test_update_user_wrong_payload(test_client, user_id, user_payload_updated):user_payload_updated["first_name"] = (True  # first_name should be a string not a boolean)response = test_client.patch(f"/api/users/{user_id}", json=user_payload_updated)assert response.status_code == 422response_json = response.json()assert response_json == {"detail": [{"type": "string_type","loc": ["body", "first_name"],"msg": "Input should be a valid string","input": True,}]}def test_update_user_doesnt_exist(test_client, user_id, user_payload_updated):response = test_client.patch(f"/api/users/{user_id}", json=user_payload_updated)assert response.status_code == 404response_json = response.json()assert response_json["detail"] == f"No User with this id: `{user_id}` found"

执行测试用例的时候，可以直接点击ide，这里使用的是macOs pycharm中的三角，也可以执行

pytest

F&Q

执行为macOS Sonoma 14.0 python3.9.6

# python3 -V
Python 3.9.6

1.执行uvicorn app.main:app --host localhost --port 8000 --reload 报错 zsh: /usr/local/bin/uvicorn: bad interpreter

1. 查看python路径

% which python3
/usr/bin/python3

2. 检修uvicorn脚本的Shebang行

which uvicorn
/Python/3.9/bin/uvicorn

既然报错解释器相关的问题，进入到文件中，将第一行解释器的路径修改为

#!/usr/bin/python3

保存退出即可。

2.生成requirement.txt时，pip3 list pipreqs有，但是执行pipreqs提示command not found

1. 查询pipreqs命令

ffastapi_curdapi % which pipreqs                    
pipreqs not found
fastapi_curdapi % pip3 list| grep pipreqs
pipreqs                   0.5.0
fastapi_curdapi % pipreqs
zsh: command not found: pipreqs
fastapi_curdapi % 
bfastapi_curdapi % which python3
/usr/bin/python3

如果which中找不到，则说明不在$PATH中，这里看到Python位于/usr/bin/下，而macOS中一般客户手动安装的路径位于/usr/local/bin/下，查询pipreqs的安装路径

fastapi_curdapi % pip3 show pipreqs | grep Location
Location: ~/Library/Python/3.9/lib/python/site-packages

重点关注输出中的Location字段，它会显示该包的安装路径。全局安装的包一般位于/usr/local/lib/pythonX.Y/site-packages（X.Y代表你的Python版本），而用户级安装的包通常位于~/Library/Python/X.Y/lib/python/site-packages
在用户级目录下发现

bin % pwd
***/Library/Python/3.9/bin
bin % ls | grep pipreqs
pipreqs

需要将对应的路径添加到PATH中，添加到用户PATH中，一般有~/.bash_profile和~/.zshrc两个文件，
~/.bash_profile 和 ~/.zshrc 是两种不同Shell的配置文件，它们分别用于Bash和Zsh这两种Unix/Linux shell环境中。它们存储着用户自定义的设置和环境变量，使得用户可以在启动shell时应用自己的个性化配置。下面是两者的具体区别：

~/.bash_profile

• 用途: 主要用于Bash shell。它是Bash在登录（login）模式下读取的第一个配置文件。当用户通过终端登录或者使用ssh远程登录时，Bash会执行这个文件来设置环境变量（如PATH）、别名（aliases）以及其他一些初始化设置。
• 特点: 适合放置那些只需要在登录时执行一次的配置，比如环境变量设置、启动程序等。
• 兼容性: 由于Bash是大多数Linux发行版和macOS的默认shell，所以这个文件在这些系统中非常常见。

~/.zshrc

• 用途: 专用于Zsh shell。无论是在交互式登录还是非登录shell中，Zsh都会读取这个文件。这意味着每次你打开一个新的Zsh终端窗口或tab时，~/.zshrc中的配置都会被应用。
• 特点: 由于Zsh提供了比Bash更多的功能和定制性，~/.zshrc可以包含更复杂的配置，如插件管理（如oh-my-zsh）、主题设置、自定义函数、别名等。
• 兼容性: 自macOS Catalina开始，Apple将Zsh设为默认shell，因此在较新的macOS系统中，~/.zshrc成为用户配置shell环境的主要文件
  如果你正在使用macOS并且主要使用的是Zsh（尤其是在Catalina及以后的版本），那么你应当关注_{/.zshrc的配置。如果你在某些场景下仍然使用Bash（或者系统默认是Bash），则需要维护}/.bash_profile。两者都非常重要，因为它们决定了你的shell环境如何响应你的命令以及提供哪些功能。在一些情况下，用户可能需要在这两个文件中都做一些配置，特别是当他们同时在Bash和Zsh环境中工作时。而且macOS Catalina是Mac的第16个主要版本，版本号为macOS 10.15,从 2019.10.8 开始，所以可以认为只需要维护~/.zshrc即可。

# 使用的macos的默认python，位于/usr/bin下，而用户级一般位于/usr/local/bin下，因此需要将用户级安装目录加入到PATH中
export PATH="自己的目录地址/Library/Python/3.9/bin:$PATH"

保存退出后，执行

source ~/.zshrc

这里大家会好奇机器明明Sonoma 14.0为什么要修改 .zshrc呢，当然是确定本机的默认Shell已经从Bash是否切换到了Zsh。执行命令

% echo $SHELL
/bin/zsh

发现这里已经切换到zsh，所以上面需要修改~/.zshrc

3.pydantic_core._pydantic_core.ValidationError: 1 validation error for UserBaseSchema

在执行pytest全量的case时报错如下：

E       pydantic_core._pydantic_core.ValidationError: 1 validation error for UserBaseSchema
E       updatedAt
E         Input should be a valid datetime [type=datetime_type, input_value=None, input_type=NoneType]
E           For further information visit https://errors.pydantic.dev/2.7/v/datetime_type../../../../Library/Python/3.9/lib/python/site-packages/pydantic/main.py:1220: ValidationError

这个错误信息表明你在使用Pydantic模型（UserBaseSchema）验证数据时遇到了问题，具体来说，错误发生在updatedAt字段，它期望一个有效的datetime类型值，但实际上收到了None，所以发现在schemas.py中定义updatedAt和createdAt都使用了初始化值为None的情况

...address: str = Noneactivated: bool = FalsecreatedAt: datetime = None# updatedAt: datetime = NoneupdatedAt: datetime = Field(default_factory=datetime.utcnow)
...