-
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/ # 配置文件目录