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

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 数据库中创建数据库 example_db：

//Login to MySQL
mysql -u root -p

//Create database named example_db
CREATE DATABASE example_db;

创建数据库表 users：

CREATE TABLE
  `users` (
    `id` int unsigned NOT NULL AUTO_INCREMENT,
    `name` varchar(255) DEFAULT NULL,
    `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (`id`)
  ) ENGINE = InnoDB DEFAULT CHARSET = utf8mb3

往数据库表中插入一条数据：

INSERT INTO users (name) VALUES ("Atom");

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 Depends
import mysql.connector


def get_db_connection():
    connection = mysql.connector.connect(
        host='localhost',
        port=3306,
        user="root",
        password="123456",
        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, Depends
from mysql.connector import cursor
from db import get_db
import json


app = FastAPI()

# def get_db(db: cursor.MySQLCursor = Depends(get_db)):
#     return db

@app.get("/users/")
async def get_users(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.fetchall()
    if result:
        return {"user_id": result[0][0], "username": result[0][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 WatchFiles
INFO:     Started server process [60503]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

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

访问 http://127.0.0.1:8000/users/，可以获取当前数据库中的所有用户数据：

"users":[[1,"Atom","2023-04-20T13:14:00"],[2,"Nuoza","2023-04-20T13:14:20"]]}

当访问 http://127.0.0.1:8000/users/1 时，结果：

{"user_id":1,"username":"Atom"}

新增一个用户，http://127.0.0.1:8000/user_name/Dole。操作如图：

总结

本文总结了通过 MySQL 驱动连接 Python， 实现了一个 FastAPI 连接 MySQL 数据库的示例， 并结合代码结构设计了目录结构和实现了获取用户和增加用户的功能，但是本文只是简单使用了原生的 MySQL，其实 FastAPI 支持结合 SQLAlchemy 更方面创建模型的数据库语句，今后会继续沿用这个示例写用户的登录和注册功能， 敬请期待。

希望本文能对你有所帮助，如果喜欢本文，可以点个关注。

下一篇文章见！宇宙古今无有穷期，一生不过须臾，当思奋争。

参考链接：

• 官方文档：SQL (Relational) Databases
• Server-side rendering with FastAPI and MySQL
• Create REST API to perform CRUD Operations using FastAPI and MySQL