Python-FastAPI框架介绍
1. FastAPI简介
FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建基于 Python 的 API。它具有简单易用的特性，同时也提供了高度自动化的文档生成功能，使得开发者可以更加高效地构建和部署 API 服务。功能强大、易于使用且高性能的 Web 框架，适用于构建各种规模的 API 服务。它的简洁语法、自动生成文档和异步支持等特性使得开发 API 更加轻松和愉快
中文文档：https://fastapi.tiangolo.com/zh/
GitHub项目地址：https://github.com/tiangolo/fastapi
本篇文章会简单的介绍一下概念，然后通过一个完整的实战项目来展开介绍各个功能模块。
FastAPI 关键特性
快速高效：基于 Starlette 和 Pydantic 构建，具有高性能和低延迟的特点，支持异步处理请求，利用 Python 的协程提高并发性能。简单易用：使用标准的 Python 类型注解来定义 API 的输入和输出参数，无需编写大量的文档和验证代码。自动生成文档：通过访问/docs 路径可以查看自动生成的交互式 API 文档，包含了每个端点的详细说明、请求和响应的模型结构以及示例请求和响应。数据验证：利用 Pydantic 提供的数据验证功能，可以自动验证请求数据的格式和类型，并进行数据转换。依赖注入：支持依赖注入，可以方便地将依赖项注入到处理函数中，例如数据库连接、配置等。类型检查：利用 Python 类型提示和 Pydantic 的数据模型，可以在开发过程中进行类型检查，减少错误和调试时间。中间件支持：支持使用中间件扩展框架的功能，例如认证、日志记录等。标准化：支持 OpenAPI 规范，可以生成符合规范的 API 文档和客户端代码。 2. FastAPI安装
安装FastAPI模块
pip install fastapipip install uvicornuvicorn用于运行 Python 的异步 Web 应用程序，与许多流行的 Python 框架（如 FastAPI、Starlette 等）兼容，可以帮助开发者构建高效的异步 Web 服务
3. FastAPI入门体验
小试牛刀demo，跑一遍官方小示例
创建一个main.py #!/usr/bin/env python3# -*- coding:utf-8 -*-# @Project :wangting_FastAPI# @File :main.py.py# @Author :wangting_666from typing import Unionfrom fastapi import FastAPIapp = FastAPI()@app.get("/")async def read_root(): return {"Hello": "World"}@app.get("/items/{item_id}")async def read_item(item_id: int, q: Union[str, None] = None): return {"item_id": item_id, "q": q} async 关键字用于定义异步函数。异步函数可以在执行过程中暂停并允许其他代码执行，直到某些条件满足后再恢复执行。在 FastAPI 中，使用 async 可以使函数能够处理异步操作，例如异步的数据库查询、IO 操作等，以提高性能和并发能力。在这个例子中，read_root 和 read_item 函数都是异步函数，它们使用了 async 关键字来定义。这样的函数可以通过 await 关键字调用其他异步函数，或者执行需要等待的异步操作，而不会阻塞整个应用程序的执行。uvicorn main:app --reload这里的 main 是你的 Python 文件名（不含扩展名main方法的意思），app 是你创建的 FastAPI 应用程序实例。
uvicorn main:app --reload 命令含义如下:
main：main.py 文件（一个 Python “模块”）。app：在 main.py 文件中通过 app = FastAPI() 创建的对象。--reload：让服务器在更新代码后重新启动。仅在开发时使用该选项。 打开浏览器并验证访问：
http://localhost:8000/
http://localhost:8000/items/1
图例1
图例2
通过以上代码创建了一个具有以下功能的 API：
通过 路径/ 和 /items/{item_id} 接受 HTTP 请求。以上 路径 都接受 GET 操作（也被称为 HTTP 方法）。/items/{item_id} 路径 有一个 路径参数 item_id 并且应该为 int 类型。/items/{item_id} 路径 有一个可选的 str 类型的 查询参数 q。 4. FastAPI交互式文档
访问 http://127.0.0.1:8000/docs 可打开API文档界面
5. FastAPI入门体验代码升级
小试牛刀demo，官方小示例迭代
main.py
from typing import Unionfrom fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel): name: str price: float is_offer: Union[bool, None] = None@app.get("/")def read_root(): return {"hello": "world"}@app.get("/item/{item_id}")def read_item(item_id: int, q: Union[str, None] = None): return {"item_id": item_id, "q": q}@app.put("/items/{item_id}")def update_item(item_id: int, item: Item): return {"item_name": item.name, "item_id": item_id}服务器将会自动重载（因为在上面的步骤中你向 uvicorn 命令添加了 --reload 选项）
由于之前在 PyCharm 的 Terminal 中运行 uvicorn 命令启动 FastAPI 应用程序，有--reload参数，所以无需重新运行，如没有增加这个参数，需要在Terminal中重新执行uvicorn
打开URLhttp://localhost:8000/docs
找到put，点击 Try it out ； 可以填写参数并直接调用 API
图例1：
点击「Execute」按钮，用户界面将会和 API 进行通信，发送参数，获取结果并在屏幕上展示
图例2：
6. FastAPI路径介绍
fastapi支持各种请求方式：
| 装饰器 | HTTP 方法 | 说明 | 使用注意事项 |
|---|---|---|---|
@app.get() | GET | 处理 HTTP GET 请求，获取资源的信息或数据。 | 应该是幂等的，不应该对服务器状态进行修改。不要包含对资源的修改操作。 |
@app.post() | POST | 处理 HTTP POST 请求，创建新资源或提交数据。 | POST 请求应该包含一个请求体，应避免包含敏感信息。 |
@app.put() | PUT | 处理 HTTP PUT 请求，更新已存在的资源或创建指定标识的资源。 | PUT 请求应该包含一个完整的资源表示，用于替换原始资源。 |
@app.patch() | PATCH | 处理 HTTP PATCH 请求，部分更新已存在的资源的内容。 | PATCH 请求应该包含一个用于指定需要更新的部分资源内容的请求体。应避免更新资源的标识或其他关键信息。 |
@app.delete() | DELETE | 处理 HTTP DELETE 请求，删除指定资源。 | DELETE 请求应该谨慎使用，应使用权限验证和确认机制。 |
@app.options() | OPTIONS | 处理 HTTP OPTIONS 请求，获取目标资源支持的通信选项。 | OPTIONS 请求通常由浏览器在跨域请求时发送，一般不需要直接处理。 |
@app.head() | HEAD | 处理 HTTP HEAD 请求，仅返回响应头信息，不返回实际内容。获取资源的响应头信息，而不获取实际的资源内容。 | HEAD 请求与 GET 请求类似，但不返回实际的资源内容。 |
@app.trace() | TRACE | 处理 HTTP TRACE 请求，追踪请求在传输链路上的路径，用于调试。 | TRACE 请求通常用于调试和诊断网络问题，一般不在业务逻辑中直接使用。 |
7. 项目实战示例1 (CRUD)
让我们一起探索一个完整的 FastAPI 项目实战！在这个项目中，帮助大家深入了解 FastAPI 框架的强大功能，并通过实际示例演示其灵活性和易用性。项目示例可以提供一个直观、易理解的教程，帮助您快速掌握 FastAPI 的核心概念和最佳实践。跟随我们一起探索，开启您的 FastAPI 之旅。通过一个魔兽世界游戏的职业信息库的增删改查需求来慢慢了解各个功能。
7-1. 项目目录结构
app0415
app ( package包，功能模块 )__init__.pylogger.pymain.pytodo.py bin （ 项目执行入口 ） run_server.py config （ 配置文件 ） config.py log （ 日志输出 ） app.log tmp （存放临时文件用）requirements.txt （ 模块依赖包清单 ） 
7-2. 准备工作
需要一个MySQL数据库-- 建测试库CREATE DATABASE `wow`;use wow;# 创建测试表CREATE TABLE `wow_info` ( `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '角色id', `role` varchar(255) DEFAULT NULL COMMENT '角色简称', `role_cn` varchar(255) DEFAULT NULL COMMENT '角色类型', `role_pinyin` varchar(255) DEFAULT NULL COMMENT '角色拼音', `zhuangbei` varchar(255) DEFAULT NULL COMMENT '装备类型', `tianfu` varchar(255) DEFAULT NULL COMMENT '天赋类型', PRIMARY KEY (`id`)) ENGINE=InnoDB AUTO_INCREMENT=667 DEFAULT CHARSET=utf8;# 插入样例数据INSERT INTO `wow_info` VALUES (1, 'fs', '法师', 'fashi', '布甲', '冰法|火法|奥法');INSERT INTO `wow_info` VALUES (2, 'ms', '牧师', 'mushi', '布甲', '神牧|戒律|暗牧');INSERT INTO `wow_info` VALUES (3, 'ss', '术士', 'shushi', '布甲', '毁灭|痛苦|恶魔');INSERT INTO `wow_info` VALUES (4, 'dz', '盗贼', 'daozei', '皮甲', '狂徒|刺杀|敏锐');INSERT INTO `wow_info` VALUES (5, 'ws', '武僧', 'wuseng', '皮甲', '酒仙|踏风|织雾');INSERT INTO `wow_info` VALUES (6, 'xd', '德鲁伊', 'xiaode', '皮甲', '恢复|平衡|野性|守护');INSERT INTO `wow_info` VALUES (7, 'dh', '恶魔猎手', 'emolieshou', '皮甲', '复仇|浩劫');INSERT INTO `wow_info` VALUES (8, 'lr', '猎人', 'lieren', '锁甲', '兽王|生存|射击');INSERT INTO `wow_info` VALUES (9, 'sm', '萨满', 'saman', '锁甲', '恢复|增强|元素');INSERT INTO `wow_info` VALUES (10, 'long', '龙人', 'longren', '锁甲', '湮灭|恩护|增辉');INSERT INTO `wow_info` VALUES (11, 'dk', '死亡骑士', 'siwangqishi', '板甲', '鲜血|冰霜|邪恶');INSERT INTO `wow_info` VALUES (12, 'zs', '战士', 'zhanshi', '板甲', '武器|狂暴|防护');INSERT INTO `wow_info` VALUES (13, 'sq', '圣骑士', 'shengqi', '板甲', '神圣|防护|惩戒');
有条件建议安装一个Postman接口调试工具，非常实用方便
pycharm开发工具和python环境7-3. 代码介绍
run_server.py
#!/usr/bin/env python3# -*- coding:utf-8 -*-# @Project :app0415# @File :run_server.py# @Time :2024/4/15 19:46# @Author :wangting_666import uvicornif __name__ == "__main__": uvicorn.run("app.main:app", host="127.0.0.1", port=8000, reload=True)todo.py
#!/usr/bin/env python3# -*- coding:utf-8 -*-# @Project :app0415# @File :todo.py# @Time :2024/4/15 19:46# @Author :wangting_666from typing import List, Optionalfrom pydantic import BaseModelimport pymysql# MySQL连接配置config = { 'host': 'wangting_host_ip', 'port': 3306, 'user': 'root', 'password': '123456', 'database': 'wow',}# 连接到MySQL数据库def connect_to_mysql(): return pymysql.connect(**config)# 定义WowInfo模型class WowInfo(BaseModel): id: int role: str role_cn: str role_pinyin: str zhuangbei: str tianfu: str# 获取所有魔兽职业信息def get_wowinfo_all() -> List[WowInfo]: try: conn = connect_to_mysql() with conn.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute("SELECT * FROM wow_info") info = [WowInfo(**row) for row in cursor.fetchall()] return info except Exception as e: print(f"查不到职业信息: {e}") return []# 获取单个魔兽职业信息def get_wowinfo(role: str) -> Optional[WowInfo]: try: conn = connect_to_mysql() with conn.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute("SELECT * FROM wow_info WHERE role = %s", (role,)) info = cursor.fetchone() return WowInfo(**info) if info else None except Exception as e: print(f"查不到职业信息: {e}") return None# 创建魔兽职业信息def create_wowinfo(wowinfo: WowInfo) -> WowInfo: try: conn = connect_to_mysql() with conn.cursor() as cursor: cursor.execute( "INSERT INTO wow_info (id,role, role_cn,role_pinyin,zhuangbei,tianfu) VALUES (%s, %s, %s,%s,%s,%s)", (wowinfo.id, wowinfo.role, wowinfo.role_cn, wowinfo.role_pinyin, wowinfo.zhuangbei, wowinfo.tianfu)) conn.commit() return wowinfo except Exception as e: print(f"创建职业信息失败: {e}") return None# 更新魔兽职业信息def update_wowinfo(id: int, wowinfo: WowInfo) -> Optional[WowInfo]: try: conn = connect_to_mysql() with conn.cursor() as cursor: cursor.execute("UPDATE wow_info SET role=%s, role_cn=%s, role_pinyin=%s,zhuangbei=%s,tianfu=%s WHERE id=%s", (wowinfo.role, wowinfo.role_cn, wowinfo.role_pinyin, wowinfo.zhuangbei, wowinfo.tianfu, id)) conn.commit() return wowinfo except Exception as e: print(f"更新职业信息失败: {e}") return None# 删除魔兽职业信息def delete_wowinfo(id: int) -> bool: try: conn = connect_to_mysql() with conn.cursor() as cursor: cursor.execute("DELETE FROM wow_info WHERE id = %s", (id,)) conn.commit() return True except Exception as e: print(f"删除职业信息失败: {e}") return Falsemain.py
#!/usr/bin/env python3# -*- coding:utf-8 -*-# @Project :app0415# @File :main.py# @Time :2024/4/15 19:46# @Author :wangting_666from fastapi import FastAPI, HTTPExceptionfrom config.config import settingsfrom app.logger import loggerfrom app.todo import WowInfo, get_wowinfo_all, get_wowinfo, create_wowinfo, update_wowinfo, delete_wowinfofrom typing import Listapp = FastAPI(title=settings.app_name)@app.get("/get_wowinfo_all/", response_model=List[WowInfo])async def read_wowinfos(): logger.info("查询所有魔兽世界角色...") return get_wowinfo_all()@app.get("/get_wowinfo/{role}", response_model=WowInfo)async def read_wowinfo(role: str): logger.info(f"查询魔兽世界角色名称为: {role}...") info = get_wowinfo(role) if info is None: logger.error(f" 角色 {role} 不存在.") raise HTTPException(status_code=404, detail="查询失败") return info@app.post("/create_wowinfo/", response_model=WowInfo)async def create_new_wowinfo(info: WowInfo): logger.info("创建魔兽世界角色") return create_wowinfo(info)@app.put("/update_wowinfo/{id}", response_model=WowInfo)async def update_existing_wowinfo(id: int, wowinfo: WowInfo): logger.info(f"更新魔兽世界角色 id {id}...") existing_info = update_wowinfo(id, wowinfo) if existing_info is None: logger.error(f"职业信息ID:{id}不存在") raise HTTPException(status_code=404, detail="职业信息ID不存在") return existing_info@app.delete("/delete_wowinfo/{id}")async def delete_existing_wowinfo(id: int): logger.info(f"删除魔兽世界角色 id {id}...") success = delete_wowinfo(id) if not success: logger.error(f"兽世界角色 id {id} 不存在") raise HTTPException(status_code=404, detail="id 不存在") return {"status": "success", "message": "删除魔兽世界角色 id {id}成功"}logger.py
#!/usr/bin/env python3# -*- coding:utf-8 -*-# @Project :app0415# @File :logger.py# @Time :2024/4/15 19:46# @Author :wangting_666import loggingfrom config.config import settingsimport oslog_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../log")os.makedirs(log_dir, exist_ok=True)logging.basicConfig(filename=os.path.join(log_dir, "app.log"), level=settings.log_level)logger = logging.getLogger(__name__)config.py
#!/usr/bin/env python3# -*- coding:utf-8 -*-# @Project :app0415# @File :config.py# @Time :2024/4/15 19:46# @Author :wangting_666from pydantic import BaseSettingsclass Settings(BaseSettings): app_name: str = "TodoApp" log_level: str = "INFO"settings = Settings()7-4. 运行服务进行验证
7-4-1. 启动服务
pycharm方式
我这里是通过pycharm方式，如果是服务器运行，只需要执行python3 /home/wangting/run_server.py运行起来即可
开始功能测试之前，可以通过上方介绍的/docs接口查看自动生成的接口文档
http://127.0.0.1:8000/docs
7-4-2. 验证服务
后续验证服务，get请求可以通过浏览器、postman、接口的/docs功能均可；我这里使用postman进行演示，也可以通过上方介绍的/docs中try it out 中进行execute进行调试验证
get_wowinfo_allget_wowinfo_all 对应了get_wowinfo_all方法，@app.get可以看出是get请求，路由是/get_wowinfo_all
@app.get("/get_wowinfo_all/", response_model=List[WowInfo])async def read_wowinfos(): logger.info("查询所有魔兽世界角色...") return get_wowinfo_all()
/get_wowinfo/{role} 对应了get_wowinfo方法，@app.get可以看出也是get请求，路由是/get_wowinfo_all/{role}
@app.get("/get_wowinfo/{role}", response_model=WowInfo)async def read_wowinfo(role: str): logger.info(f"查询魔兽世界角色名称为: {role}...") info = get_wowinfo(role) if info is None: logger.error(f" 角色 {role} 不存在.") raise HTTPException(status_code=404, detail="查询失败") return info
create_wowinfo 对应了create_wowinfo方法，@app.post可以看出是post请求，路由是/create_wowinfo/
但与get请求不同的是，post请求需要提供body请求体
{ "id": 666, "role": "create_role", "role_cn": "create_role_cn", "role_pinyin": "create_role_pinyin", "zhuangbei": "create_zhuangbei", "tianfu": "create_tianfu"}@app.post("/create_wowinfo/", response_model=WowInfo)async def create_new_wowinfo(info: WowInfo): logger.info("创建魔兽世界角色") return create_wowinfo(info)
此时查看一下MySQL数据库，验证是否真的写入到数据库
update_wowinfo/{id} 对应了update_wowinfo方法，@app.put可以看出是put请求，路由是/update_wowinfo/{id}
与post请求一样，也需要提供body请求体
{ "id": 666, "role": "update_role", "role_cn": "update_role_cn", "role_pinyin": "update_role_pinyin", "zhuangbei": "update_zhuangbei", "tianfu": "update_tianfu"}@app.put("/update_wowinfo/{id}", response_model=WowInfo)async def update_existing_wowinfo(id: int, wowinfo: WowInfo): logger.info(f"更新魔兽世界角色 id {id}...") existing_info = update_wowinfo(id, wowinfo) if existing_info is None: logger.error(f"职业信息ID:{id}不存在") raise HTTPException(status_code=404, detail="职业信息ID不存在") return existing_info
查询数据库数据，id=666的数据内容已经被修改
MariaDB [wow]> select * from wow_info where id=666 \G;*************************** 1. row *************************** id: 666 role: update_role role_cn: update_role_cnrole_pinyin: update_role_pinyin zhuangbei: update_zhuangbei tianfu: update_tianfudelete_wowinfo/{id} 对应了delete_wowinfo方法，@app.delete可以看出是delete请求，路由是/delete_wowinfo/{id}
@app.delete("/delete_wowinfo/{id}")async def delete_existing_wowinfo(id: int): logger.info(f"删除魔兽世界角色 id {id}...") success = delete_wowinfo(id) if not success: logger.error(f"兽世界角色 id {id} 不存在") raise HTTPException(status_code=404, detail="id 不存在") return {"status": "success", "message": "删除魔兽世界角色 id {id}成功"}
此时查询数据库数据id=666的数据内容已经不存在了
MariaDB [wow]> select * from wow_info where id=666 \G;Empty set (0.000 sec)7-4-3. 项目实战总结
这个 FastAPI 项目展示了一个完整的后端应用程序，用于管理魔兽世界角色的信息。通过这个项目，我们深入了解了 FastAPI 框架的使用，以及如何构建 RESTful API。
首先，我们学习了如何定义数据模型，使用 Pydantic 创建了 WowInfo 模型来表示魔兽世界角色的信息。然后，我们编写了一系列处理 CRUD 操作的函数，用于获取、创建、更新和删除角色信息。这些函数与 MySQL 数据库进行交互，通过 pymysql 库执行 SQL 查询和操作。
接下来，我们使用 FastAPI 创建了一个基于路由的 Web 服务，定义了不同的端点来处理不同的 HTTP 请求。通过最常用的装饰器 @app.get、@app.post、@app.put 和 @app.delete，我们将每个端点与对应的处理函数绑定在一起。
最后，我们使用了日志记录功能和配置设置，使项目更具可维护性和可扩展性。我们将日志记录到文件中，并使用配置类来管理应用程序的设置。
这个项目为我们提供了一个完整的 FastAPI 实战示例，展示了如何使用 FastAPI 框架构建一个简单但功能完备的后端应用程序。通过这个示例，我们不仅学习了 FastAPI 框架的核心概念和基本用法，还掌握了与数据库交互、路由定义、日志记录等高级功能。
8. 项目实战示例2 (前后端)
8-1. 项目目录结构
app0416 server_monitor_frontend index.html server_monitor_backend.py示例是一个前后端协作，仅展示逻辑demo，没有加入太多的功能模块
在 app0416 目录下有一个名为 server_monitor_backend.py 的文件，用于启动后端 FastAPI 服务器。在同一目录下还有一个名为 server_monitor_frontend 的文件夹，其中包含一个名为 index.html 的文件，用于显示服务器监控系统的界面。如果需要自定义样式，可以在 server_monitor_frontend 文件夹中创建一个名为 style.css 的文件，并在 index.html 中引入，此目录主要用来开发前端功能。最终实现打开http://127.0.0.1:8000/，可以5秒刷新展示一次系统的服务器资源情况。
8-2. 代码介绍
server_monitor_backend.py#!/usr/bin/env python3# -*- coding:utf-8 -*-# @Project :app0416# @File :server_monitor_backend.py.py# @Time :2024/4/16 19:58# @Author :wangting_666from fastapi import FastAPIfrom fastapi.staticfiles import StaticFilesfrom starlette.responses import HTMLResponse, FileResponseapp = FastAPI()# 将静态文件目录设置为 server_monitor_frontend 目录app.mount("/static", StaticFiles(directory="server_monitor_frontend"), name="static")@app.get("/", response_class=HTMLResponse)async def get_index_page(): return FileResponse("server_monitor_frontend/index.html")@app.get("/stats")async def get_server_stats(): # Your server stats logic here return {"cpu_percent": 50, "memory_percent": 60, "disk_percent": 70}if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)index.html
<!DOCTYPE html><html><head> <title>服务器资源监控</title> <style> body { font-family: Arial, sans-serif; text-align: center; } </style></head><body><h1>Server Monitor</h1><div id="stats"></div><script> function fetchStats() { fetch('http://127.0.0.1:8000/stats') .then(response => response.json()) .then(data => { document.getElementById('stats').innerHTML = ` <p><h3>CPU使用率: ${data.cpu_percent}%</h3></p> <p><h3>内存使用率: ${data.memory_percent}%</h3></p> <p><h3>磁盘使用率: ${data.disk_percent}%</h3></p> `; }) .catch(error => console.error('Error fetching stats:', error)); } fetchStats(); setInterval(fetchStats, 5000); // Fetch stats every 5 seconds</script></body></html>8-3. 运行服务进行验证
运行server_monitor_backend.py
打开URL ： http://127.0.0.1:8000/
8-4. 前后端示例总结
这个案例演示了如何使用 FastAPI 构建一个简单的服务器监控系统。后端 FastAPI 提供了 /stats 路由，用于获取服务器的监控数据，而前端界面通过 JavaScript 定时获取数据并动态显示在页面上。这个示例结合了后端和前端技术，展示了如何利用 FastAPI 快速构建 API，以及如何使用 JavaScript 实现动态页面交互。