FastAPI 实战：高效构建 Web API 并集成 MySQL 数据库

2025年11月14日 互联网

FastAPI 实战：高效构建 Web API 并集成 MySQL 数据库

FastAPI 作为一款现代化的 Python Web 框架，凭借其高性能、易用性和自动生成 API 文档的特性，迅速成为开发 Web API 的热门选择。对于需要持久化数据的项目，连接 MySQL 数据库是不可或缺的一环。本文将详细介绍如何使用 FastAPI 快速开发 Web API 项目，并重点讲解如何与 MySQL 数据库建立连接，实现数据的增删改查（CRUD）操作。

一、FastAPI 简介与优势

FastAPI 基于 Starlette 和 Pydantic 构建，支持异步请求处理，能够提供接近原生异步框架的性能。其核心优势包括：

  1. 自动生成交互式 API 文档：基于 OpenAPI 和 JSON Schema，自动生成 Swagger UI 和 ReDoc 文档，极大提升开发效率。
  2. 类型提示支持：利用 Python 的类型提示功能，提供编辑器支持，减少错误。
  3. 高性能：基于 Starlette 的异步特性，能够处理高并发请求。
  4. 简洁的语法：代码简洁易读，适合快速开发。

二、环境准备与依赖安装

在开始开发前，需要确保环境配置正确，并安装必要的依赖库。

1. 创建项目目录

  1. mkdir fastapi_mysql_demo
  2. cd fastapi_mysql_demo
  3. python -m venv venv
  4. source venv/bin/activate  # Linux/macOS
  5. # venv\Scripts\activate  # Windows

2. 安装 FastAPI 和 Uvicorn

Uvicorn 是一个 ASGI 服务器，用于运行 FastAPI 应用。

  1. pip install fastapi uvicorn

3. 安装 MySQL 连接库

选择 mysql-connector-python 或 pymysql 作为 MySQL 的 Python 客户端。

  1. pip install mysql-connector-python
  2. # 或
  3. pip install pymysql

三、连接 MySQL 数据库

1. 配置数据库连接

在项目中创建一个 database.py 文件，用于管理数据库连接。

  1. import mysql.connector
  2. from mysql.connector import Error
  4. def create_connection():
  5.     """创建并返回数据库连接"""
  6.     try:
  7.         connection = mysql.connector.connect(
  8.             host='localhost',
  9.             user='your_username',
  10.             password='your_password',
  11.             database='your_database'
  12.         )
  13.         if connection.is_connected():
  14.             print("成功连接到 MySQL 数据库")
  15.             return connection
  16.     except Error as e:
  17.         print(f"连接 MySQL 数据库时出错: {e}")
  18.         return None

2. 使用连接池（可选）

对于高并发应用，建议使用连接池管理数据库连接。

  1. from mysql.connector import pooling
  3. dbconfig = {
  4.     "host": "localhost",
  5.     "user": "your_username",
  6.     "password": "your_password",
  7.     "database": "your_database"
  8. }
  10. connection_pool = pooling.MySQLConnectionPool(
  11.     pool_name="my_pool",
  12.     pool_size=5,
  13.     **dbconfig
  14. )
  16. def get_connection():
  17.     """从连接池获取连接"""
  18.     return connection_pool.get_connection()

四、实现 CRUD 操作

1. 创建数据模型

使用 Pydantic 定义数据模型，用于数据的验证和序列化。

  1. from pydantic import BaseModel
  3. class Item(BaseModel):
  4.     name: str
  5.     description: str | None = None
  6.     price: float
  7.     on_offer: bool = False

2. 实现 CRUD 路由

在 main.py 中定义 FastAPI 应用，并实现 CRUD 路由。

  1. from fastapi import FastAPI, HTTPException
  2. from typing import List
  3. from database import create_connection, get_connection  # 假设已实现
  4. from pydantic import BaseModel
  6. app = FastAPI()
  8. # 模拟数据库表操作
  9. class Item(BaseModel):
  10.     name: str
  11.     description: str | None = None
  12.     price: float
  13.     on_offer: bool = False
  15. # 假设的数据库表
  16. items_db = []
  18. @app.post("/items/", response_model=Item)
  19. def create_item(item: Item):
  20.     """创建新项目"""
  21.     # 实际应用中，这里应插入数据库
  22.     items_db.append(item)
  23.     return item
  25. @app.get("/items/", response_model=List[Item])
  26. def read_items():
  27.     """获取所有项目"""
  28.     # 实际应用中，这里应从数据库查询
  29.     return items_db
  31. @app.get("/items/{item_id}", response_model=Item)
  32. def read_item(item_id: int):
  33.     """根据ID获取项目"""
  34.     # 实际应用中，这里应从数据库查询
  35.     if item_id < len(items_db):
  36.         return items_db[item_id]
  37.     raise HTTPException(status_code=404, detail="Item not found")
  39. @app.put("/items/{item_id}", response_model=Item)
  40. def update_item(item_id: int, item: Item):
  41.     """更新项目"""
  42.     # 实际应用中，这里应更新数据库
  43.     if item_id < len(items_db):
  44.         items_db[item_id] = item
  45.         return item
  46.     raise HTTPException(status_code=404, detail="Item not found")
  48. @app.delete("/items/{item_id}")
  49. def delete_item(item_id: int):
  50.     """删除项目"""
  51.     # 实际应用中，这里应从数据库删除
  52.     if item_id < len(items_db):
  53.         del items_db[item_id]
  54.         return {"message": "Item deleted successfully"}
  55.     raise HTTPException(status_code=404, detail="Item not found")

实际数据库操作示例（以 create_item 为例）：

  1. from fastapi import FastAPI, HTTPException
  2. from pydantic import BaseModel
  3. from database import get_connection  # 假设使用连接池
  5. app = FastAPI()
  7. class Item(BaseModel):
  8.     name: str
  9.     description: str | None = None
  10.     price: float
  11.     on_offer: bool = False
  13. @app.post("/items/", response_model=Item)
  14. def create_item(item: Item):
  15.     """创建新项目并插入数据库"""
  16.     connection = get_connection()
  17.     if connection is None:
  18.         raise HTTPException(status_code=500, detail="Database connection failed")
  20.     try:
  21.         cursor = connection.cursor()
  22.         query = """
  23.         INSERT INTO items (name, description, price, on_offer)
  24.         VALUES (%s, %s, %s, %s)
  25.         """
  26.         cursor.execute(query, (item.name, item.description, item.price, item.on_offer))
  27.         connection.commit()
  28.         # 假设能获取自增ID（实际需根据数据库调整）
  29.         item.id = cursor.lastrowid  
  30.         return item
  31.     except Error as e:
  32.         connection.rollback()
  33.         raise HTTPException(status_code=500, detail=str(e))
  34.     finally:
  35.         cursor.close()
  36.         connection.close()  # 注意：实际连接池实现中可能不需要显式关闭

五、运行与测试

1. 运行 FastAPI 应用

  1. uvicorn main:app --reload

2. 测试 API

使用浏览器访问 http://127.0.0.1:8000/docs，利用 Swagger UI 测试 API。

或使用 curl：

  1. # 创建项目
  2. curl -X POST "http://127.0.0.1:8000/items/" -H "accept: application/json" -H "Content-Type: application/json" -d '{"name": "Foo", "description": "An item", "price": 10.5, "on_offer": false}'
  4. # 获取所有项目
  5. curl "http://127.0.0.1:8000/items/"

六、最佳实践与注意事项

  1. 错误处理：妥善处理数据库异常，避免泄露敏感信息。
  2. 连接管理：使用连接池管理数据库连接，避免频繁创建和销毁连接。
  3. 安全性：使用环境变量存储数据库凭据，避免硬编码。
  4. 异步支持：对于 I/O 密集型操作，考虑使用异步数据库客户端（如 asyncpg 配合 databases 库）。
  5. ORM 使用：对于复杂项目，可考虑使用 SQLAlchemy 或 Tortoise-ORM 等 ORM 工具。

七、总结

通过本文，我们了解了如何使用 FastAPI 快速开发 Web API 项目，并详细讲解了如何与 MySQL 数据库建立连接，实现数据的 CRUD 操作。FastAPI 的高性能和易用性，结合 MySQL 的稳定性和可靠性，为开发高效的 Web API 项目提供了强大的支持。希望本文能为你的 FastAPI 开发之路提供有益的指导。

最新文章