企业官网建设流程全解析

一. 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, table=True): """ 图书模型 参数说明： - table=True: 标记这是一个数据库表模型 - Field(): SQLModel 专用的字段定义工具 """ #数据库主键字段 id: Optional[int] = Field( default=None, # 新建时不需要提供，数据库自动生成 primary_key=True, # 标记为主键 description="图书 ID" ) #标题字段 title: str = Field( index=True, # 创建索引，加速查询 nullable=False, # 不允许为空 description="图书标题" ) #作者字段 anthor : str = Field( description="作者名称" ) #价格字段 price: float = Field( gt = 0, #价格要大于0 description="图书价格" ) # 描述字段（可选） description: Optional[str] = Field( default=None, description="图书描述" )

1.4 配置数据库链接

首先先介绍几个重要的概念：

1.4.1 create_async_engine —— 连接数据库的 "总管"

1. 是什么：SQLAlchemy 提供的创建异步引擎的函数，返回值是 AsyncEngine 对象（所以在写代码的时候一定要记住使用一个变量来接住这个创建的对象，用来后续的操作）。它不直接执行你的业务逻辑，而是所有数据库操作的底层通道。（类似于银行的大厅总管），也是用来连接数据库的操作。

2. 作用：

• 连接池管理：维护一组到数据库的物理连接并反复复用。新建一个数据库连接开销很大（握手、鉴权），连接池让高频请求不必每次都新建，这是性能的关键。（相当于银行窗口，不是你新来一个客户，就给你新建造一个窗口，而是每个窗口都在复用，大大提高了资源利用）
• 异步执行 SQL：把 ORM 层的操作翻译成 SQL，交给异步驱动（asyncpg、aiomysql、aiosqlite）发往数据库，全程 async/await，不阻塞事件循环。
• 统一配置入口：echo=True 打印 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_commit=False)，一行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, table=True): # ① SQLModel 定义模型（表结构） id: int | None = Field(default=None, primary_key=True) name: str engine = create_async_engine("sqlite+aiosqlite:///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 # sqlite+aiosqlite:/// 表示使用异步 SQLite 驱动 DATABASE_URL = "sqlite+aiosqlite:///./books.db" # 1. 创建异步引擎 engine = create_async_engine( DATABASE_URL, echo=True, # 打印 SQL 语句（开发时有用，生产环境建议关闭），开启后程序每执行一条SQL语句，都会完整打印在日志里。 future=True # 启用 SQLAlchemy 2.0 特性（记住就行） ) """在开发阶段它非常实用： 你用 ORM 写的是 Python 代码（比如`select(User).where(User.id == 1)`），最终会被翻译成原生 SQL 发给数据库。开启`echo=True`就能直观看到： - 生成的 SQL 逻辑对不对、有没有生成多余的查询； - 联表、条件写法有没有问题，能不能命中索引； - 事务提交、回滚的时机是否符合预期。 是开发阶段调错、优化 SQL 的神器。 #### 2. 为啥生产环境建议关闭？ 三个核心原因，越靠后越重要： 1. **损耗性能** 打印日志本质是磁盘 IO 操作，本身速度很慢。生产环境每秒可能有上百上千次数据库请求，每条都打印会大量消耗 CPU 和 IO 资源，直接拖慢接口响应速度，是很典型的非必要性能损耗。 2. **泄露敏感数据** SQL 语句里会携带真实的业务参数，比如用户手机号、密码哈希、身份证号、订单金额等。这些数据明文打印在日志里，一旦日志泄露或者被未授权人员访问，会造成严重的数据安全问题，不符合数据合规要求。 3. **日志冗余爆炸** 生产环境请求量极大，全量打印 SQL 会让日志文件极速膨胀，既占用服务器磁盘空间，又会把真正的错误日志淹没在海量 SQL 里，出问题排查的时候反而找不到重点。""" # 2. 创建异步 Session 工厂 # expire_on_commit=False: 提交后对象仍然可用 async_session = async_sessionmaker( engine, # 绑定异步引擎，会话从它的连接池拿数据库连接 class_=AsyncSession, # 指定会话类型为异步会话（同步会话是 Session），必须与异步引擎匹配。 # 默认情况下，会话提交（commit）后，查询出来的对象会 “过期”（无法再访问属性）， # 设为 False 后，提交后对象仍可正常使用（开发更友好） expire_on_commit=False # ) # 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 编程指南》", author="Guido van Rossum", price=68.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查询

常用的查询方法如下：

一句话记法：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 books

2.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 book

2.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 True

2.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 编程指南》", author="Guido van Rossum", price=68.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 版）》", author="Guido van Rossum", price=78.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🎉 所有测试通过！")