-
guest说:
FastAPI单元测试讲解实例，以学生管理为例
目录
1、准备工作
在开始编写测试代码之前，我们需要准备一些必要的工作：
-
安装
pytest要进行FastAPI的单元测试，我们需要安装
pytest，它是一款功能强大的Python测试框架，支持测试用例的编写、执行以及结果的收集，所以我们需要先安装它：$ pip install pytest -
安装
uvicornuvicorn是一款异步非阻塞的Python Web服务器，它支持ASGI协议，所以我们可以使用它来启动我们的FastAPI应用，并用于进行测试：$ pip install uvicorn -
编写FastAPI应用
为了方便讲解，我们编写一个简单的学生管理的FastAPI应用：
# main.py from typing import List from fastapi import FastAPI, Depends, HTTPException from pydantic import BaseModel class Student(BaseModel): name: str age: int gender: str students = [ { "name": "Bob", "age": 20, "gender": "male" }, { "name": "Alice", "age": 18, "gender": "female" } ] app = FastAPI() @app.get("/students/") async def get_students(): return students @app.post("/students/") async def create_student(student: Student): students.append(student.dict()) return student @app.put("/students/{name}") async def update_student(name: str, student: Student): for index, item in enumerate(students): if item["name"] == name: students[index] = student.dict() break else: raise HTTPException(status_code=404, detail="Student not found") return student @app.delete("/students/{name}") async def delete_student(name: str): for index, item in enumerate(students): if item["name"] == name: del students[index] break else: raise HTTPException(status_code=404, detail="Student not found") return {"msg": "delete success"}
2、编写测试代码
2.1 测试获取学生列表
首先，我们来测试获取学生列表的接口，编写测试用例：
# test_main.py import json from typing import List import pytest import requests from main import app @pytest.fixture(scope="module") def test_client(): test_client = app.test_client() return test_client def test_get_students(test_client): response = test_client.get("/students/") data = json.loads(response.data) assert len(data) == 2 students = [ { "name": "Bob", "age": 20, "gender": "male" }, { "name": "Alice", "age": 18, "gender": "female" } ] assert data == students测试用例：
- 先通过
pytest.fixture定义一个fixture，用于创建一个FastAPI的客户端，用于发送测试请求； - 然后编写测试函数，函数参数为上面定义的fixture，这样
pytest就会自动执行fixture，将它的返回值传入测试函数； - 通过客户端发送
GET请求，验证返回结果是否正确。
2.2 测试创建学生
接下来，我们来测试创建学生的接口，编写测试用例：
# test_main.py import json import pytest from main import app @pytest.fixture(scope="module") def test_client(): test_client = app.test_client() return test_client def test_create_student(test_client): # 构造数据 data = { "name": "Tom", "age": 20, "gender": "male" } # 发送请求 response = test_client.post("/students/", json=data) # 验证结果 assert response.status_code == 200 data = json.loads(response.data) assert data == data测试用例：
- 通过
pytest.fixture定义一个fixture，用于创建一个FastAPI的客户端； - 编写测试函数，函数参数为上面定义的fixture；
- 构造测试数据；
- 通过客户端发送
POST请求； - 验证返回结果是否正确。
2.3 测试更新学生信息
接下来，我们来测试更新学生信息的接口，编写测试用例：
# test_main.py import json import pytest from main import app @pytest.fixture(scope="module") def test_client(): test_client = app.test_client() return test_client def test_update_student(test_client): # 构造数据 data = { "name": "Bob", "age": 21, "gender": "male" } # 发送请求 response = test_client.put("/students/Bob", json=data) # 验证结果 assert response.status_code == 200 data = json.loads(response.data) assert data == data测试用例：
- 通过
pytest.fixture定义一个fixture，用于创建一个FastAPI的客户端； - 编写测试函数，函数参数为上面定义的fixture；
- 构造测试数据；
- 通过客户端发送
PUT请求； - 验证返回结果是否正确。
2.4 测试删除学生
最后，我们来测试删除学生的接口，编写测试用例：
# test_main.py import json import pytest from main import app @pytest.fixture(scope="module") def test_client(): test_client = app.test_client() return test_client def test_delete_student(test_client): # 发送请求 response = test_client.delete("/students/Bob") # 验证结果 assert response.status_code == 200 data = json.loads(response.data) assert data == {"msg": "delete success"}测试用例：
- 通过
pytest.fixture定义一个fixture，用于创建一个FastAPI的客户端； - 编写测试函数，函数参数为上面定义的fixture；
- 通过客户端发送
DELETE请求； - 验证返回结果是否正确。
3、编写测试执行程序
-
-
guest说:
举例讲解下FastAPI的单元测试，以学生信息管理为例，举例讲解下原理和使用方法
- 单元测试的原理：
单元测试是指分离的测试，它的目的是检查每个独立的单元是否正确工作，单元测试可以帮助检测单元之间的相互依赖，从而确保程序的稳定性和正确性。
- 使用FastAPI进行单元测试：
（1）安装pytest，并在项目根目录创建 tests/ 目录；
（2）在 tests/ 目录下创建一个 test_students.py 文件，用来编写学生信息管理的测试用例；
（3）使用 pytest 在 tests/ 目录下执行 test_students.py 文件，检查测试用例是否正确；
（4）如果测试用例通过，则代表学生信息管理的功能是正常工作的。
-
-
guest说:
我要做一个纯后端的API接口项目，文章信息管理系统，用到的框架是FastAPI，使用的模块有sqlalchemy、pydantic、logging，请帮我列出合理的目录结构及文件，给出目录和文件的注释，并给出每个文件内的代码编写，最后打包供下载下来
部署上线。
目录结构：
- config/ # 配置文件目录
- config.py # 配置文件，存放数据库和其他环境配置信息
- doc/ # 文档目录
- README.md # 项目概述文档
- logs/ # 日志目录
- access.log # 访问日志
- error.log # 错误日志
- src/ # 主模块目录
- init.py # 模块初始化文件
- app.py # 启动入口文件
- models.py # 数据模型文件
- routes/ # 路由文件目录
- init.py # 路由文件初始化文件
- articles.py # 文章路由文件
- utils/ # 工具函数目录
- init.py # 工具函数初始化文件
- db.py # 操作数据库工具函数
- requirements.txt # 依赖库文件
- .gitignore # git忽略文件
# 配置文件，存放数据库和其他环境配置信息 import os # 数据库配置 DB_USERNAME = os.getenv("DB_USERNAME") DB_PASSWORD = os.getenv("DB_PASSWORD") DB_HOST = os.getenv("DB_HOST") DB_PORT = os.getenv("DB_PORT") DB_NAME = os.getenv("DB_NAME")# 启动入口文件 import uvicorn from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from src.models import Base from src.routes.articles import router from src.utils.db import init_db # 初始化app app = FastAPI( title="文章信息管理系统", description="纯后端的API接口项目，用于管理文章信息", version="1.0.0", ) # 跨域设置 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 注册路由 app.include_router(router, prefix="/api/v1") # 初始化数据库 init_db(Base) # 启动服务 if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)# 数据模型文件 from sqlalchemy import Column, Integer, String from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class Article(Base): __tablename__ = "articles" id = Column(Integer, primary_key=True, index=True) title = Column(String) content = Column(String)routes/articles.py
# 文章路由文件 from fastapi import APIRouter, Depends from sqlalchemy.orm import Session from src.models import Article from src.utils.db import get_db router = APIRouter() @router.get("/") def get_articles(db: Session = Depends(get_db)): articles = db.query(Article).order_by(Article.id).all() return articles @router.post("/") def create_article(article: Article, db: Session = Depends(get_db)): db.add(article) db.commit() db.refresh(article) return article @router.put("/{article_id}") def update_article(article_id: int, article: Article, db: Session = Depends(get_db)): db.query(Article).filter(Article.id == article_id).update( { "title": article.title, "content": article.content, } ) db.commit() return article @router.delete("/{article_id}") def delete_article(article_id: int, db: Session = Depends(get_db)): article = db.query(Article).get(article_id) db.delete(article) db.commit() return {"msg": "success"}utils/db.py
# 操作数据库工具函数 from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from src.config import DB_USERNAME, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME # 数据库连接 engine = create_engine( f"mysql+pymysql://{DB_USERNAME}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4" ) # 数据库Session SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 初始化数据库 def init_db(Base): Base.metadata.create_all(bind=engine) # 获取数据库Session def get_db(): db = SessionLocal() try: yield db finally: db.close()requirements.txt
fastapi>=0.61.0 uvicorn>=0.11.2 SQLAlchemy>=1.3.16 pydantic>=1.6.1 PyMySQL>=0.9.3.gitignore
# 日志文件 logs/* - config/ # 配置文件目录
