一. 什么是ORMORMObject-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] aiomysql2、创建数据库在MySQL中创建数据库MySQL 语句/关键字不区分大小写。首先保证自己已经安装了MySQL并在环境变量里添加了相关路径。打开 cmd在命令行中输入mysql -u root -p输入密码后进入 MySQL。使用以下命令创建数据库CREATE DATABASE 数据库名;使用以下命令查看数据库是否创建成功show databases;3、创建数据库引擎数据库引擎所需的 URL 是根据实际使用的数据库、用户名密码、IP地址、端口和数据库名拼写出来的数据库 URL 的拼写规则为mysqlaiomysql://用户名:密码IP地址:端口/数据库名MySQL数据库url地址写法be likeDATABASE_URL mysqlaiomysql://用户名:密码数据库IP:3306/数据库名PostgreSQLurl地址写法be likeDATABASE_URL postgresqlasyncpg://user:password127.0.0.1:5432/dbnameSQLite数据库url地址写法be likeDATABASE_URL sqliteaiosqlite:///./test.db实际代码from sqlalchemy.ext.asyncio import create_async_engine # 创建异步引擎 DATABASE_URL mysqlaiomysql://root:123456localhost:3306/fastapi01?charsetutf8mb4 async_engine create_async_engine( DATABASE_URL, # 数据库URL echoTrue, # 输出SQL日志 pool_size10, # 连接池中长期保留的连接数量相当于一直待命的连接 max_overflow20 # 连接池临时多开的连接数量用完会释放 )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_defaultfunc.now(), defaultfunc.now, # 注意这里只传递函数 comment创建时间, ) update_time: Mapped[datetime.datetime] mapped_column( DateTime, insert_defaultfunc.now(), defaultfunc.now, # 注意这里只传递函数 onupdatefunc.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 # 创建表——定义模型类 class Book(Base): __tablename__ book id: Mapped[int] mapped_column(primary_keyTrue, 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 启动时自动执行建表函数确保数据库表在接口使用前已经准备好。代码如下# 启动时自动建表 # 定义建表函数 # 注意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(lifespanlifespan)点击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 # 2.根据自己使用的数据库构造数据库URL ASYNC_DATABASE_URL mysqlaiomysql://root:123456localhost:3306/fastapi01?charsetutf8mb4 # 3.构造数据库引擎并把构造好的URL传递进去 async_engine create_async_engine( ASYNC_DATABASE_URL, echoTrue, pool_size10, max_overflow20, ) # 4.创建表——定义基类 class Base(DeclarativeBase): create_time: Mapped[datetime.datetime] mapped_column( DateTime, insert_defaultfunc.now(), defaultfunc.now, comment创建时间, ) update_time: Mapped[datetime.datetime] mapped_column( DateTime, insert_defaultfunc.now(), defaultfunc.now, onupdatefunc.now(), comment修改时间, ) # 4.创建表——定义模型类 class Book(Base): __tablename__ book id: Mapped[int] mapped_column(primary_keyTrue, 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(lifespanlifespan) app.get(/) async def root(): return {message: Hello World}