ORM概述

9.1 ORM 介绍

对象关系映射（Object-Relational Mapping, ORM）技术是现代软件架构中解决持久层管理的核心工具。它通过在应用程序的对象模型与关系数据库的表结构之间建立一座桥梁，显著提高了开发效率并增强了代码的可维护性。

flowchart LR
    A[程序中的对象] --> B[ORM 框架]
    B --> C[生成 SQL 语句]
    C --> D[数据库]
    D --> E[返回结果]
    E --> F[ORM 将结果转换为对象]
    F --> A

简单来说，ORM 就像是一个翻译官，它在你的类对象和数据库表之间建立了一座桥梁。

通过定义一个类来映射一张表，对类对象的操作就是对表的操作。

ORM 的主要优势

减少重复的 SQL 代码
自动处理数据库连接和事务
代码更简洁易读
自动防止 SQL 注入攻击
业务逻辑与数据访问逻辑分离

9.2 SQLAlchemy介绍

SQLAlchemy 是 Python 中最强大、最流行的 ORM（对象关系映射）工具包，它提供了完整的 ORM 功能以及灵活的 SQL 表达式语言。

I. pip安装

安装sqlalchemy和aiomysql模块

pip install sqlalchemy[asyncio] aiomysql

II. 基本使用

创建数据库,使用数据库工具连接Mysql数据库

create database orm_db;

配置数据库文件，创建db.py文件
- 都是固定写法，实际开发可以直接复制修改配置信息

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker

# 数据库UR
ASYNC_DATABASE_URL = "mysql+aiomysql://root:root@localhost:3306/orm_db?charset=utf8mb4"
# 创建异步引擎
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,  # 可选：输出SQL日志
    pool_size=10,  # 设置连接池中保持的持久连接数
    max_overflow=20  # 设置连接池允许创建的额外连接数
)
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# 依赖项，用于获取数据库会话
async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

定义模型类，创建model.py文件

# 导入字段类型
from sqlalchemy import Column,String,Integer
from sqlalchemy.orm import  DeclarativeBase

# 1. 声明 ORM 模型的基类
# 方式1： 采用类形式，官方推荐。当有多个模型类需要使用共有属性和方法是可以定义在Base类中
class Base(DeclarativeBase):
    pass
# 方式2：
# Base = declarative_base()

# 2. 定义User用户类
class User(Base):
    __tablename__ = "users"
    # 
    id = Column(Integer, primary_key=True, index=True)
    phone = Column(String(255), unique=True, index=True)
    username = Column(String(100))

建表，在main.py中指定

from fastapi import FastAPI

# 导入异步建表
from sqlalchemy.ext.asyncio import create_async_engine
from models import Base

app = FastAPI()

# 定义异步建表函数
async def create_tables():
    async_engine = create_async_engine("mysql+aiomysql://root:root@localhost:3306/orm_db?charset=utf8mb4")  # 替换为你的异步数据库 URL
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# 将建表函数加载到fastapi服务中
@app.on_event("startup")
async def startup_event():
    await create_tables()

# ... 其余代码

注意: 建表的操作也可以直接在mysql数据库通过SQL语句实现
1
2
3
4
5
6
7
create  table if not exists  orm_db.users
(
    id  int auto_increment primary key,
    phone    varchar(255) null,
    username varchar(100) null,
    unique (phone)
);
两种建表方式都可以

特性 SQL建表 SQLAlchemy建表

方式直接执行SQL 通过Python代码生成SQL

可读性 对DBA和熟悉SQL的人更直观对Python开发者更友好

维护性 需要手动维护SQL脚本表结构作为Python代码，与应用代码一起维护

灵活性 可以使用数据库所有特性受限于SQLAlchemy的功能，但大部分常用特性都支持

迁移管理 通常借助第三方工具（如Flyway）可使用Alembic进行迁移，与SQLAlchemy集成良好

数据库兼容性 针对特定数据库，换数据库需重写SQL 可支持多种数据库，但需注意不同数据库的兼容性

路由匹配中使用orm，在main.py中增加编写如下代码

from db import get_database
# 导入异步操作类
from sqlalchemy.ext.asyncio import AsyncSession
# 导入select查询方法
from sqlalchemy import select
# 导入自己定义的User类
from models import User


@app.get("/user/")
async def read_user(db: AsyncSession = Depends(get_database)):
    # 异步获取Users表中的所有数据
    result = await db.execute(select(User)) # 执行后返回一个orm对象
    user = result.scalars().all()

    return user

特性	SQL建表	SQLAlchemy建表
方式	直接执行SQL	通过Python代码生成SQL
可读性	对DBA和熟悉SQL的人更直观	对Python开发者更友好
维护性	需要手动维护SQL脚本	表结构作为Python代码，与应用代码一起维护
灵活性	可以使用数据库所有特性	受限于SQLAlchemy的功能，但大部分常用特性都支持
迁移管理	通常借助第三方工具（如Flyway）	可使用Alembic进行迁移，与SQLAlchemy集成良好
数据库兼容性	针对特定数据库，换数据库需重写SQL	可支持多种数据库，但需注意不同数据库的兼容性