一. ORM概念ORM(Object-Relational Mapping对象关系映射)是一种技术它允许开发者按照面向对象编程语言来操作关系型数据库1.1 核心思想类(Class)对应一个数据库表Table对象对应表中的一行属性对应表中的一列比如一个学生类对应着一个学生信息表学生张三对应着张三在这表中的一行信息学生的学号这个属性就对应着这个表中的id这一列1.2 工具选择SQLmodel安装依赖# 安装 SQLModel包含 SQLAlchemy 和 Pydantic pip install sqlmodel # 安装异步 SQLite 驱动 pip install aiosqlite # 安装异步 Pytest-asyncio 插件 pip install pytest-asyncio # 如果使用清华源 pip install sqlmodel aiosqlite -i https://pypi.tuna.tsinghua.edu.cn/simple/ pip install fastmcp -i https://pypi.tuna.tsinghua.edu.cn/simple/1.3 定义一个model类型代码示例from typing import Optional from sqlmodel import SQLModel,Field class Book(SQLModel, tableTrue): 图书模型 参数说明 - tableTrue: 标记这是一个数据库表模型 - Field(): SQLModel 专用的字段定义工具 #数据库主键字段 id: Optional[int] Field( defaultNone, # 新建时不需要提供数据库自动生成 primary_keyTrue, # 标记为主键 description图书 ID ) #标题字段 title: str Field( indexTrue, # 创建索引加速查询 nullableFalse, # 不允许为空 description图书标题 ) #作者字段 anthor : str Field( description作者名称 ) #价格字段 price: float Field( gt 0, #价格要大于0 description图书价格 ) # 描述字段可选 description: Optional[str] Field( defaultNone, description图书描述 )1.4 配置数据库链接首先先介绍几个重要的概念1.4.1 create_async_engine —— 连接数据库的 总管1. 是什么SQLAlchemy 提供的创建异步引擎的函数返回值是 AsyncEngine 对象所以在写代码的时候一定要记住使用一个变量来接住这个创建的对象用来后续的操作。它不直接执行你的业务逻辑而是所有数据库操作的底层通道。类似于银行的大厅总管也是用来连接数据库的操作。2. 作用连接池管理维护一组到数据库的物理连接并反复复用。新建一个数据库连接开销很大握手、鉴权连接池让高频请求不必每次都新建这是性能的关键。相当于银行窗口不是你新来一个客户就给你新建造一个窗口而是每个窗口都在复用大大提高了资源利用异步执行 SQL把 ORM 层的操作翻译成 SQL交给异步驱动asyncpg、aiomysql、aiosqlite发往数据库全程 async/await不阻塞事件循环。统一配置入口echoTrue 打印 SQL 日志、pool_size 控制连接池大小、pool_pre_ping 防止失效连接等都在这里配。1.4.2 AsyncSession —— 操作数据库的 上下文 / 工作台1. 是什么异步版本的数据库会话类。一次会话就是一次 与数据库的对话通常对应一个事务。1.4.3 sessionmaker —— 生产会话的 工厂1. 是什么创建会话工厂的函数。它返回一个可复用的 会话生成器调用一次Session()就产出一个AsyncSession实例。2. 作用集中配置绑定哪个引擎、expire_on_commit、autoflush等默认参数只写一次全应用共享。引擎就是最开始的create_async_engine这个创造的就是引擎避免重复代码不用在每处手写AsyncSession(engine, expire_on_commitFalse)一行Session()即可。全局单例典型做法是模块里建一个Session sessionmaker(engine, class_AsyncSession)然后依赖注入到路由 / 服务里。类比会话的 自动售货机—— 配置一次投币规则之后每次按一下就出一个会话1.4.4 配合流程from sqlmodel import SQLModel, Field, create_async_engine # ④① from sqlalchemy.ext.asyncio import AsyncSession # ② from sqlalchemy.orm import sessionmaker # ③ class User(SQLModel, tableTrue): # ① SQLModel 定义模型表结构 id: int | None Field(defaultNone, primary_keyTrue) name: str engine create_async_engine(sqliteaiosqlite:///app.db) # ② 创建引擎连接池 Session sessionmaker(engine, class_AsyncSession) # ③ 创建会话工厂 async def main(): async with Session() as session: # ④ 从工厂拿会话 session.add(User(name张三)) # 会话内操作 await session.commit() # 提交事务一句话总结SQLModel 定结构 → 引擎管连接 → 工厂造会话 → 会话做增删改查四者合起来构成异步数据库操作的完整链路记住这个流程先创建引擎用来连接→再创建会话工厂→在初始化DB下面是示例代码from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlmodel import SQLModel # 数据库连接 URL # sqliteaiosqlite:/// 表示使用异步 SQLite 驱动 DATABASE_URL sqliteaiosqlite:///./books.db # 1. 创建异步引擎 engine create_async_engine( DATABASE_URL, echoTrue, # 打印 SQL 语句开发时有用生产环境建议关闭开启后程序每执行一条SQL语句都会完整打印在日志里。 futureTrue # 启用 SQLAlchemy 2.0 特性记住就行 ) 在开发阶段它非常实用 你用 ORM 写的是 Python 代码比如select(User).where(User.id 1)最终会被翻译成原生 SQL 发给数据库。开启echoTrue就能直观看到 - 生成的 SQL 逻辑对不对、有没有生成多余的查询 - 联表、条件写法有没有问题能不能命中索引 - 事务提交、回滚的时机是否符合预期。 是开发阶段调错、优化 SQL 的神器。 #### 2. 为啥生产环境建议关闭 三个核心原因越靠后越重要 1. **损耗性能** 打印日志本质是磁盘 IO 操作本身速度很慢。生产环境每秒可能有上百上千次数据库请求每条都打印会大量消耗 CPU 和 IO 资源直接拖慢接口响应速度是很典型的非必要性能损耗。 2. **泄露敏感数据** SQL 语句里会携带真实的业务参数比如用户手机号、密码哈希、身份证号、订单金额等。这些数据明文打印在日志里一旦日志泄露或者被未授权人员访问会造成严重的数据安全问题不符合数据合规要求。 3. **日志冗余爆炸** 生产环境请求量极大全量打印 SQL 会让日志文件极速膨胀既占用服务器磁盘空间又会把真正的错误日志淹没在海量 SQL 里出问题排查的时候反而找不到重点。 # 2. 创建异步 Session 工厂 # expire_on_commitFalse: 提交后对象仍然可用 async_session async_sessionmaker( engine, # 绑定异步引擎会话从它的连接池拿数据库连接 class_AsyncSession, # 指定会话类型为异步会话同步会话是 Session必须与异步引擎匹配。 # 默认情况下会话提交commit后查询出来的对象会 “过期”无法再访问属性 # 设为 False 后提交后对象仍可正常使用开发更友好 expire_on_commitFalse # ) # 3. 初始化数据库表结构 async def init_db(): 创建所有表 这个函数应该在应用启动时调用一次 # 通过异步引擎开启一个事务连接engine.begin() 会自动管理事务退出上下文时提交 # conn 是数据库连接实例。 async with engine.begin() as conn: # 如果需要重建表可以先删除 # await conn.run_sync(SQLModel.metadata.drop_all) # 创建所有表如果不存在 await conn.run_sync(SQLModel.metadata.create_all)#conn.run_sync(...)把同步的 create_all 方法适配到异步连接中执行 SQLModel.metadata.create_all 扫描 metadata 中所有已注册的模型比如 Book 根据模型定义生成对应的 CREATE TABLE SQL 语句 连接数据库执行这些 SQL创建表如果表已存在则跳过不会重复创建。二.增删改查操作示例2.1 create新增信息这个就是create的流程图我们在后端发送一个post请求然后api接口就会创建一个session类似于这样 async with async_session() as session:这个async_session()就是上面异步session工厂创造出来的操作对象之后通过这个session操作窗口来操作先session.add把要新增的信息预保存之后再异步await提交也就是await session.commit提交到数据库就相当于执行了一下insert语句之后我们在执行await session.refresh()刷新对象把数据库里的对象拿出来这就是create的流程了下面是代码示例# CRUD实战————新增图书函数 async def create_book(book_data: Book) - Book: 创建新图书 参数 book_data: Book对象不含id 返回 包含id的Book对象 async with async_session() as session: # 添加会话不做IO所以不加await这个操作把对象标记为待保存 session.add(book_data) # 提交事务写入数据库执行insert这个SQL语句 await session.commit() #刷新对象从数据库里重新加载对象获取自增ID await session.refresh(book_data) return book_data现在我们得到了一个create的操作函数我们要怎么测试呢这时候我们的pytest就派上用场了写上import pytest 引入 Python 生态最主流的自动化测试框架 pytest用来给你的代码尤其是你正在写的数据库操作、业务逻辑编写并运行自动化测试验证功能是否正确、有没有隐藏 bug。之后再写上pytest.mark.asyncio 它是 pytest-asyncio 插件规定的标准标记写法在默认配置下异步测试函数必须加这个装饰器才能正常运行。这个也是测试上面函数的地方用来测试功能后面就可以测试了接下来是代码示例import pytest async def test_book_crud_operations(): 测试 Book 模型的完整 CRUD 操作 # 初始化数据库 await init_db() print(✅ 数据库初始化完成) # 测试 CREATE print(\n--- 测试创建图书 ---) book_data Book( title《Python 编程指南》, authorGuido van Rossum, price68.0, description一本关于 Python 编程的权威指南 ) created_book await create_book(book_data) print(f✅ 创建成功: ID{created_book.id}, 标题{created_book.title}) assert created_book.id is not None #python中的断言assert如果后面的为True就什么都不发生否则就会报错用来验证操作执行正确与否的2.2 Read查询常用的查询方法如下方法适用场景示例session.get(Model, id)通过主键查单条session.get(Book, 1)select(Model)查询所有记录select(Book)select(Model).where()条件查询select(Book).where(Book.price 50)select(Model).limit()限制返回条数select(Book).limit(10)一句话记法get 查主键select 查一切where 加条件limit 限条数offset 做分页。记住代码中需要有 import sqlmodel from select接下来是代码示例#查询所有图书 async def get_books() - list[Book]: 查询所有图书 返回: Book 对象列表 async with async_session() as session: # 1. 构建 SELECT 语句 statement select(Book) # 2. 执行查询 result await session.execute(statement) # 3. 获取所有结果,使用result的all方法,直接调用 .all() 得到的是 Row 行对象列表每个元素是 (Book实例,) 形式的元组而不是直接的 Book 对象列表。 books result.all() return books #按照ID来查询图书因为一个id对应一本书所以返回的是单独的Book对象 async def get_book_by_id(book_id: int) - Optional[Book]: 根据Id查询图书 参数 book_id: int图书的id 返回 Book对象或者None async with async_session() as session: #使用session.get()方法直接通过主键查询 book await session.get(Book,book_id) return book #条件查询 async def get_book_by_author(name: str) - list[Book]: 根据作者名称来查询图书 参数 作者名称name 返回 图书列表因为一个作者可能有很多书所以返回列表 async with async_session() as session: statement select(Book).where(Book.author.contains(name)) result await session.execute(statement) # 用 scalars() 提取 ORM 实体再转列表 books result.scalars().all() return books2.3 update 修改这就是修改的流程。修改数据的逻辑先查询 - 修改属性 - 保存接下来是代码示例#更新操作 async def update_book(book_id: int, new_data: Book) - Optional[Book]: 更新图书信息 参数: book_id: 要更新的图书 ID new_data: 新的图书数据 返回: 更新后的 Book 对象如果不存在返回 None async with async_session() as session: # 1. 查询要更新的图书 book await session.get(Book, book_id) if not book: return None # 2. 更新属性 book.title new_data.title book.author new_data.author book.price new_data.price if new_data.description: book.description new_data.description # 3. 添加到会话已 attached 的对象可以省略 session.add(book) # 4. 提交更改 await session.commit() # 5. 刷新对象 await session.refresh(book) return book2.4 delete删除直接代码示例async def delete_book(book_id: int) - bool: 删除图书 参数: book_id: 要删除的图书 ID 返回: True 表示删除成功False 表示图书不存在 async with async_session() as session: # 1. 查询要删除的图书 book await session.get(Book, book_id) if not book: return False # 2. 删除对象 await session.delete(book) # 3. 提交事务 await session.commit() return True2.5 完整代码from typing import Optional from orm_product.database import init_db, async_session from orm_product.model import Book #先从同一个包下引入进两个东西一个是初始化数据库的函数另外一个是数据类的内容毕竟一个类就代表一个表 import pytest #引入 Python 生态最主流的自动化测试框架 pytest用来给你的代码尤其是你正在写的数据库操作、业务逻辑编写并运行自动化测试验证功能是否正确、有没有隐藏 bug。 from sqlmodel import select # CRUD实战————新增图书函数 async def create_book(book_data: Book) - Book: 创建新图书 参数 book_data: Book对象不含id 返回 包含id的Book对象 async with async_session() as session: # 添加会话不做IO所以不加await这个操作把对象标记为待保存 session.add(book_data) # 提交事务写入数据库执行insert这个SQL语句 await session.commit() #刷新对象从数据库里重新加载对象获取自增ID await session.refresh(book_data) return book_data #查询所有图书 async def get_books() - list[Book]: 查询所有图书 返回: Book 对象列表 async with async_session() as session: # 1. 构建 SELECT 语句 statement select(Book) # 2. 执行查询 result await session.execute(statement) # 3. 获取所有结果,使用result的all方法,直接调用 .all() 得到的是 Row 行对象列表每个元素是 (Book实例,) 形式的元组而不是直接的 Book 对象列表。 books result.all() return books #按照ID来查询图书因为一个id对应一本书所以返回的是单独的Book对象 async def get_book_by_id(book_id: int) - Optional[Book]: 根据Id查询图书 参数 book_id: int图书的id 返回 Book对象或者None async with async_session() as session: #使用session.get()方法直接通过主键查询 book await session.get(Book,book_id) return book #条件查询 async def get_book_by_author(name: str) - list[Book]: 根据作者名称来查询图书 参数 作者名称name 返回 图书列表因为一个作者可能有很多书所以返回列表 async with async_session() as session: statement select(Book).where(Book.author.contains(name)) result await session.execute(statement) # 用 scalars() 提取 ORM 实体再转列表 books result.scalars().all() return books #更新操作 async def update_book(book_id: int, new_data: Book) - Optional[Book]: 更新图书信息 参数: book_id: 要更新的图书 ID new_data: 新的图书数据 返回: 更新后的 Book 对象如果不存在返回 None async with async_session() as session: # 1. 查询要更新的图书 book await session.get(Book, book_id) if not book: return None # 2. 更新属性 book.title new_data.title book.author new_data.author book.price new_data.price if new_data.description: book.description new_data.description # 3. 添加到会话已 attached 的对象可以省略 session.add(book) # 4. 提交更改 await session.commit() # 5. 刷新对象 await session.refresh(book) return book #删除操作 async def delete_book(book_id: int) - bool: 删除图书 参数: book_id: 要删除的图书 ID 返回: True 表示删除成功False 表示图书不存在 async with async_session() as session: # 1. 查询要删除的图书 book await session.get(Book, book_id) if not book: return False # 2. 删除对象 await session.delete(book) # 3. 提交事务 await session.commit() return True pytest.mark.asyncio #它是 pytest-asyncio 插件规定的标准标记写法在默认配置下异步测试函数必须加这个装饰器才能正常运行。这个也是测试上面函数的地方用来测试功能 async def test_book_crud_operations(): 测试 Book 模型的完整 CRUD 操作 # 初始化数据库 await init_db() print(✅ 数据库初始化完成) # 测试 CREATE print(\n--- 测试创建图书 ---) book_data Book( title《Python 编程指南》, authorGuido van Rossum, price68.0, description一本关于 Python 编程的权威指南 ) created_book await create_book(book_data) print(f✅ 创建成功: ID{created_book.id}, 标题{created_book.title}) assert created_book.id is not None #python中的断言assert如果后面的为True就什么都不发生否则就会报错用来验证操作执行正确与否的 # 测试 READ查询所有 print(\n--- 测试查询所有图书 ---) books await get_books() print(f✅ 共查询到 {len(books)} 本图书) assert len(books) 0 # 测试 READ按 ID 查询 print(\n--- 测试按 ID 查询 ---) found_book await get_book_by_id(created_book.id) print(f✅ 查询成功: {found_book.title}) assert found_book is not None # 测试 READ按 作者名称 查询 print(\n--- 测试按 作者名称 查询 ---) books await get_book_by_author(van) print(f✅ 查询成功:包含关键字’van‘的作者有{len(books)}本书) print(books) # 测试 UPDATE print(\n--- 测试更新图书 ---) update_data Book( title《Python 编程指南第 2 版》, authorGuido van Rossum, price78.0, description更新版的 Python 编程权威指南 ) updated_book await update_book(created_book.id, update_data) print(f✅ 更新成功: 新标题{updated_book.title}, 新价格{updated_book.price}) assert updated_book.price 78.0 # 测试 DELETE print(\n--- 测试删除图书 ---) delete_result await delete_book(created_book.id) print(f✅ 删除成功) assert delete_result is True # 验证删除后不存在 deleted_book await get_book_by_id(created_book.id) assert deleted_book is None print(✅ 确认图书已被删除) print(\n 所有测试通过)
