FastAPI整合SQLAlchemy ORM实战：连接MySQL数据库与CRUD接口开发

在前文我们已经探讨过使用ORM操作数据库的优势，本章将结合 FastAPI 与 SQLAlchemy 框架，通过一个完整的示例来演示如何进行数据库集成与基础操作，帮助大家巩固这一在现代Python后端开发中的核心技能。

环境配置与依赖安装

首先，需要安装 MySQL 的 Python 驱动程序。

pipenv install pymysql

接着，安装 SQLAlchemy ORM 库。

pipenv install SQLAlchemy

配置数据库连接

在 main.py 文件中设置数据库连接。以下配置仅为示例，在实际项目中，建议将数据库连接字符串等配置信息放入配置文件（如 .env 文件）中动态管理，而非硬编码。

# -*- coding:utf-8 –*-
from fastapi import FastAPI
from sqlalchemy import create_engine

HOST = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = '123456'
DATABASE = 'fastapi'

db_url = f"mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}"
engine = create_engine(db_url)

初始化数据库会话

通过 SQLAlchemy 的 sessionmaker 创建数据库会话工厂，并生成一个可用的会话实例。

from sqlalchemy.orm import sessionmaker

# 假设 engine 已通过上一步创建
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = SessionFactory()

定义数据模型基类

所有 ORM 模型类都将继承自这个通过 declarative_base() 生成的基类。

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

构建实体模型

定义一个 User 模型类，它映射到数据库中的 user 表。这是SQLAlchemy声明式模型定义的标准写法。

from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__: str = "user"

    id = Column(Integer, primary_key=True)
    username = Column('username', String, unique=True, index=True)
    password = Column('password', String)

实现基础的CRUD操作

以下示例在 FastAPI 应用中创建了几个简单的端点，分别对应增、改、删操作。

app = FastAPI()

# 新增用户
@app.get("/user/add")
async def user_add():
    user1 = User(username="rebort", password="123")
    session.add(user1)
    session.commit()
    return user1.id

# 修改用户信息
@app.get("/user/update")
async def user_update():
    user1 = session.query(User).filter(User.id == 1).first()
    user1.password = "456"
    session.commit()
    return True

# 删除用户
@app.get("/user/delete")
async def user_delete():
    user1 = session.query(User).filter(User.id == 1).first()
    session.delete(user1)
    session.commit()
    return True

完整代码示例

将上述步骤整合，完整的 main.py 文件内容如下：

# -*- coding:utf-8 –*-
from fastapi import FastAPI
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 数据库连接配置
HOST = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = '123456'
DATABASE = 'fastapi'

db_url = f"mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}"
engine = create_engine(db_url)

# 创建会话
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = SessionFactory()

# 声明基类
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__: str = "user"
    id = Column(Integer, primary_key=True)
    username = Column('username', String, unique=True, index=True)
    password = Column('password', String)

# 创建FastAPI应用实例
app = FastAPI()

# 新增用户接口
@app.get("/user/add")
async def user_add():
    user1 = User(username="rebort", password="123")
    session.add(user1)
    session.commit()
    return user1.id

# 修改用户接口
@app.get("/user/update")
async def user_update():
    user1 = session.query(User).filter(User.id == 1).first()
    user1.password = "456"
    session.commit()
    return True

# 删除用户接口
@app.get("/user/delete")
async def user_delete():
    user1 = session.query(User).filter(User.id == 1).first()
    session.delete(user1)
    session.commit()
    return True