一. 什么是ORM？

ORM（Object-Relational Mapping，对象关系映射）是一种用于操作数据库的编程技术，用来在面向对象编程语言与关系型数据库之间建立映射关系。

通过 ORM，开发者可以使用 Python 对象的方式操作数据库，而无需直接编写 SQL 语句。

举个例子，如果现在要创建一个名为 book 的表，可以直接在 python 代码中这么写：

class Book(Base):
__tablename__ = "book"

相当于 SQL 语句：

CREATE TABLE book (...);

使用 ORM 的优势：减少重复 SQL 代码；代码更简洁、可读性更高；提高开发效率；自动管理数据库连接和事务；使用参数化机制，降低 SQL 注入风险。

Python 中常见的 ORM 工具有：SQLAlchemy ORM（最常用、最灵活）、Django ORM（适合 Django 框架）和 Tortoise ORM（异步 ORM）。

后续内容使用的是：SQLAlchemy ORM（社区完善、功能强大、应用最广）

二、快速开始

1、安装ORM

可以直接使用以下命令在 terminal 里面安装：

pip install sqlalchemy[asyncio] aiomysql

2、创建数据库

在MySQL中创建数据库（MySQL 语句/关键字不区分大小写）。首先保证自己已经安装了MySQL并在环境变量里添加了相关路径。打开 cmd，在命令行中输入：

mysql -u root -p

输入密码后进入 MySQL。使用以下命令创建数据库：

CREATE DATABASE 数据库名;

使用以下命令查看数据库是否创建成功：

show databases;

3、创建数据库引擎

数据库引擎所需的 URL 是根据实际使用的数据库、用户名密码、IP地址、端口和数据库名拼写出来的，数据库 URL 的拼写规则为：

mysql+aiomysql://用户名:密码@IP地址:端口/数据库名

MySQL数据库url地址写法be like：

DATABASE_URL = "mysql+aiomysql://用户名:密码@数据库IP:3306/数据库名"

PostgreSQLurl地址写法be like：

DATABASE_URL = "postgresql+asyncpg://user:password@127.0.0.1:5432/dbname"

SQLite数据库url地址写法be like：

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

实际代码：

from sqlalchemy.ext.asyncio import create_async_engine

# 创建异步引擎
DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi01?charset=utf8mb4"
async_engine = create_async_engine(
    DATABASE_URL,      # 数据库URL
    echo=True,         # 输出SQL日志
    pool_size=10,      # 连接池中长期保留的连接数量，相当于一直待命的连接
    max_overflow=20    # 连接池临时多开的连接数量，用完会释放
)

代码中 async_engine 是 create_async_engine() 创建出来的异步数据库引擎对象。

4、创建表

4.1、定义基类

先有基类，基类放通用的属性和字段（放所有表都需要的属性/字段）。

import datetime
from sqlalchemy import DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# 创建表——定义基类
class Base(DeclarativeBase):
    create_time: Mapped[datetime.datetime] = mapped_column(
        DateTime,
        insert_default=func.now(),
        default=func.now,            # 注意这里只传递函数
        comment="创建时间",
    )
    update_time: Mapped[datetime.datetime] = mapped_column(
        DateTime,
        insert_default=func.now(),
        default=func.now,            # 注意这里只传递函数
        onupdate=func.now(),
        comment="修改时间",
    )

DeclarativeBase 让类具备 被 ORM 映射为数据库表 的能力。而在 Base 中定义了两个字段：create_time 和 update_time，这两个字段相当于两个默认字段，凡是 Base 的子类都会具备这两个字段。以 create_time 为例，通过结构：

字段名:Mapped[python层面的数据类型] = mapped_column(
    数据库层面的数据类型,
    insert_default = 数据库层面要填入的默认值,
    default = python层面的默认值,
    comment = "对这个字段的说明"
    )

来定义一个字段，可以看到其实 ORM 像是一个“翻译”，它提供了 Mapped 类型标注和 mapped_column() 函数，既负责 python 层面的注释，也负责数据库层面的注释，相当于同时告诉 python 和数据库这个字段是什么类型的，它的默认值是什么。

但是为什么要同时在 python 层面和数据库层面同时进行注释呢？这是因为如果只在 Python 层定义 default，那么通过 ORM 插入数据是正常的，但如果直接使用 SQL 操作数据库，默认值不会生效，可能导致数据为空；如果只在数据库层定义 insert_default，那么无论 ORM 还是 SQL 插入，数据库中的数据都是完整的，但在 ORM 操作过程中，Python 对象在提交前可能无法获取该字段的值。

而 sqlalchemy 提供了 DateTime 和 func 等数据库层面的数据类型和操作函数。值得注意的一点是：在 default 中只传递了函数 func.now，而非函数运行的结果 func.now()，这是因为如果写成 default = func.now() 的话，default 的值就被固定为程序启动/加载时的时间，所以，以后不管新增多少数据，create_time 或者 update_time 都是是同一个固定值。

4.2、定义模型类（每一个具体表格都是一个模型类）

模型类就是要定义的具体表格（每一个具体表格都是一个模型类）。比如现在创建一个名为book的表格：

from sqlalchemy import String, Float, Integer

# 创建表——定义模型类
class Book(Base):
    __tablename__ = 'book'

    id: Mapped[int] = mapped_column(Integer, primary_key=True, comment='书籍id')
    bookname: Mapped[str] = mapped_column(String(255), comment='书名')
    author: Mapped[str] = mapped_column(String(255), comment='作者')
    price: Mapped[float] = mapped_column(Float, comment='价格')
    publisher: Mapped[String] = mapped_column(String(255), comment='出版社')

由上述代码可知，模型类的定义跟基类定义的格式非常相似。在模型类定义中，表名用 __tablename__ 字段定义，参数 primary_key 可设置指定字段为主键，String 是 sqlalchemy 提供的数据库层面的数据类型，String(255) 意味着数据库字段类型是字符串（VARCHAR），最大长度为 255 个字符。

5、启动时自动建表

当所需要的表格已经定义好，最后需要通过 lifespan 机制，在 FastAPI 启动时自动执行建表函数，确保数据库表在接口使用前已经准备好。代码如下：

from contextlib import asynccontextmanager
# 启动时自动建表

# 定义建表函数
# 注意Base.metadata.create_all后面没有小括号
async def create_tables():                               
    async with async_engine.begin() as conn:            # 异步资源管理(异步打开+异步关闭)
        await conn.run_sync(Base.metadata.create_all)    

# 定义FastAPI启动时异步执行建表函数create_tables()
@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_tables()
    yield

# 将上面自定义的lifespan函数传递给FastAPI的lifespan函数（FastAPI有一个自己的lifespan函数）
app = FastAPI(lifespan=lifespan)

在 create_tables() 函数中，用async with异步管理资源，async_engien 是之前的异步数据库引擎对象实例，async_engine.begin() 用来获取一个数据库连接，这个连接命名为 conn。run_sync() 函数可以让 create_all 这个同步函数在异步环境中执行。Base.metadata.create_all 拆开看：Base是之前定义好的基类，metadata 可理解为 Base 名下所有表结构的“登记册”，里面记录着：Book 表叫什么名字、有哪些字段、字段类型是什么、主键是谁、备注是什么，而 create_all 则是把登记册里的所有表，真正创建到数据库里。注意，这里 create_all 后面不带小括号，create_all 后面不加括号，是因为要把函数本身交给 run_sync() 去调用，而不是现在立刻执行。

@asynccontextmanager 是 Python 标准库 contextlib  提供的装饰器，用来把一个普通的异步函数，变成可以配合 async with 使用的资源管理函数。在这种函数中，yield 前面的代码会在进入时执行，yield 后面的代码会在退出时执行，因此常用于程序启动和关闭时要做的事情。比如在 FastAPI 中，常用它来定义应用生命周期逻辑：yield 前执行启动任务（如连接数据库、建表），yield 后执行关闭任务（如释放资源、关闭连接）。

点击run查看有无报错，运行成功会出现如下提示：

如果没报错可以看看 pycharm 里面的 database 里的表格情况：

填写好 user 和 password，点击 ok。就能在右侧边栏的 Database 中看到在 fastapi01 里面创建的 book 表格了：

6、完整代码

# 1.导包
import datetime
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import DateTime, func, String, Float, Integer

# 2.根据自己使用的数据库构造数据库URL
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi01?charset=utf8mb4"

# 3.构造数据库引擎，并把构造好的URL传递进去
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,
    pool_size=10,
    max_overflow=20,
)

# 4.创建表——定义基类
class Base(DeclarativeBase):
    create_time: Mapped[datetime.datetime] = mapped_column(
        DateTime,
        insert_default=func.now(),
        default=func.now,
        comment="创建时间",
    )
    update_time: Mapped[datetime.datetime] = mapped_column(
        DateTime,
        insert_default=func.now(),
        default=func.now,
        onupdate=func.now(),
        comment="修改时间",
    )

# 4.创建表——定义模型类
class Book(Base):
    __tablename__ = "book"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255), comment="书名")
    author: Mapped[str] = mapped_column(String(255), comment="作者")
    price: Mapped[float] = mapped_column(Float, comment="价格")
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

# 5.启动时自动建表——连接数据库 → 执行ORM的建表操作
async def create_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# 5.启动时自动建表——定义FastAPI启动和关闭时执行的代码
@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_tables()
    yield

# 5.启动时自动建表——将上面自定义的lifespan赋给FastAPI中特定的lifespan函数
app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root():
    return {"message": "Hello World"}
# 数据库 # python # pycharm # 后端 # sql
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

昇腾 0 Day 支持 Hunyuan4-preview，770B MoE 旗舰模型开源即适配

avatar AtomGit开源社区
cover

把 Agent 装进桌面：dsh-desktop 入驻 AtomGit

avatar AtomGit开源社区
cover

AtomCode 上线 MiMo-V2.5 系列｜限时 7 天免费体验！

avatar AtomGit开源社区