FastAPI 快速开发 Web API 项目: 连接 MySQL 数据库

2023-04-19
中国香港
本文字数：2721 字
阅读完需：约 9 分钟

FastAPI 快速开发 Web API 项目学习笔记：

第一篇：通过 Python FastAPI 开发一个快速的 Web API 项目
第二篇：FastAPI 的路由介绍与使用
第三篇：FastAPI 开发中数据校验利器 Pydantic 介绍与集成使用
第四篇：FastAPI 快速开发 Web API 项目: 路径参数和查询参数
第五篇：从 Flask 转换到 FastAPI（翻译）
第六篇：FastAPI 快速开发 Web API 项目: 响应模型与错误处理
第七篇：FastAPI 快速开发 Web API 项目: 模板和 Jinja 介绍

介绍

FastAPI 不要求我们强制使用 SQL（关系）数据库。因此，我们可以根据需求使用任何关系数据库，例如：

PostgreSQL
MySQL
SQLite
Oracle
Microsoft SQL Server
...

这篇文章以 MySQL 为例，首先需要为 MySQL 安装 Python 库，FastAPI 需要使用 Python 的 MySQL 客户端库来连接到 MySQL 数据库，这些驱动包括 mysql-connector-python 和 pymysql。

安装这些库：

pip install mysql-connector-python pymysql

复制代码

用于定义 FastAPI 依赖项的 Python 文件名没有特定的命名约定。通常，最好使用与应用程序的功能和角色匹配的名称命名，包含定义 FastAPI 依赖项的函数的 Python 文件。

在终端中，运行以下命令以在 MySQL 数据库中创建数据库：

//Login to MySQLmysql -u root -p
//Create database named example_dbCREATE DATABASE example_db;

复制代码

db.py 文件

例如，通常命名一个包含依赖项函数的 Python 文件，该函数定义与上面示例中所示的 MySQL 数据库的连接，以指示它具有与数据库相关的功能，例如 db.py 或 database.py。

此外，如果您有多个定义 FastAPI 依赖项的函数，或者如果您为不同功能定义了依赖项，则可以通过为每个功能指定不同的名称来提高代码的可读性。

例如，请考虑以下文件名：

db.py：包含定义与 MySQL 数据库连接的依赖项的文件。
auth.py：定义与身份验证相关的依赖项的文件。
validation.py：定义验证输入数据的依赖项的文件。

根据项目的大小和结构，您还可以设计目录和模块的层次结构。通过将包含定义依赖项的函数的 Python 文件放置在适当的目录中并将它们组织为模块，可以改进代码管理和可维护性。

将 MySQL 客户端库添加到 FastAPI 依赖项：使用 FastAPI 依赖项将 MySQL 客户端库添加到您的应用程序。例如，如果使用 mysql-connector-python，请按如下方式添加依赖项：

$ touch routers/db.py

复制代码

然后在 db.py 中写入如下代码：

from fastapi import Dependsimport mysql.connector
def get_db_connection():    connection = mysql.connector.connect(        host="localhost",        user="root",        password="",        database="example_db"    )    return connection
def get_db():    connection = get_db_connection()    db = connection.cursor()    try:        yield db    finally:        db.close()        connection.close()

复制代码

db_router 文件

在路由器上使用 MySQL 数据库：要将 MySQL 数据库与 FastAPI 的路由器一起使用，请使用上面定义的 get_db() 函数作为依赖项。下面是在 FastAPI 的路由器中使用 MySQL 数据库的示例。

在将 MySQL 数据库与 FastAPI 路由器一起使用的示例 Python 文件名中，通常最好根据应用程序的功能和角色对其进行命名。你可以想到这样的文件名：

main.py：包含示例代码的文件，该示例是应用程序的主要入口点，定义 FastAPI 路由器并使用 MySQL 数据库。
router.py：定义 FastAPI 路由器并包含使用 MySQL 数据库的示例的代码的文件。
db.py：包含用于连接和查询 MySQL 数据库的函数的文件。

根据项目的大小和结构，您还可以设计目录和模块的层次结构。例如，您可以通过创建名为路由器的目录，在其中创建定义 FastAPI 路由器的模块，并在其中包含使用 MySQL 数据库的示例代码来提高代码组织和可读性。

例如，可以考虑以下文件名和目录结构：

main.py：作为应用程序主入口点的文件，导入并使用路由器目录中的路由器模块。
routers/db_router.py：定义使用 MySQL 数据库的示例路由器的模块。
routers/db.py：定义用于连接和查询 MySQL 数据库的函数的模块。

db_router.py 文件写入如下内容：

from fastapi import FastAPI, Dependsfrom mysql.connector import cursorfrom db import get_dbimport json

app = FastAPI()
def get_db(db: cursor.MySQLCursor = Depends(get_db)):    return db  @app.get("/users/")async def get_user(db: cursor.MySQLCursor = Depends(get_db)):    query = "SELECT * FROM users"    db.execute(query)    result = db.fetchall()    if result:        return {"users":result}    else:        return {"error": "User not found"}      @app.get("/users/{user_id}")async def get_user(user_id: int                   , db: cursor.MySQLCursor = Depends(get_db)):    query = "SELECT * FROM users WHERE id = %s"    db.execute(query, (user_id,))    result = db.fetchone()    if result:        return {"user_id": result[0], "username": result[1]}    else:        return {"error": "User not found"}
@app.get("/user_name/{user_name}")async def insert_user(user_name:str                      ,db: cursor.MySQLCursor = Depends(get_db)):    query = "INSERT INTO users (name) VALUES (%s)"    db.execute(query, (user_name,))    result = db.fetchone()    db.execute("COMMIT")    return {"user_name":user_name}

复制代码

启动 FastAPI 服务

uvicorn db_router:app --reload
INFO:     Will watch for changes in these directories: ['/Users/sharland/dev/fastapi/MySQL/routers']INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)INFO:     Started reloader process [60501] using WatchFilesINFO:     Started server process [60503]INFO:     Waiting for application startup.INFO:     Application startup complete.

复制代码

当您在浏览器中访问 http://127.0.0.1:8000/docs 时，将显示一个屏幕，允许您使用 GUI 操作快速 API 服务。