### 如何在 FastAPI 中使用 SQLModel 执行原生 SQL 查询并通过 Pydantic 模型序列化输出 SQLAlchemy 是一种流行的 Python 数据库工具，而 SQLModel 是基于 SQLAlchemy 和 Pydantic 构建的一个现代框架。它允许开发者轻松执行原生 SQL 查询并利用 Pydantic 的强大功能来处理数据验证和序列化。 以下是实现这一目标的具体方式： #### 1. 配置数据库连接 为了能够运行原生 SQL 查询，首先需要设置数据库引擎以及会话管理器。可以参考以下代码片段完成此操作[^2]： ```python from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker DATABASE_URL = "mysql+aiomysql://user:password@host:port/dbname" engine = create_async_engine(DATABASE_URL, echo=True) async_session = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False) async def get_db(): async with async_session() as session: yield session ``` 上述代码创建了一个异步数据库引擎 `create_async_engine` 并设置了异步会话工厂 `sessionmaker` 来生成新的会话实例。 --- #### 2. 定义 SQLModel 类 SQLModel 提供了一种简单的方式来定义表结构并与之交互。下面是一个简单的例子展示如何定义一个模型类[^3]: ```python from sqlmodel import Field, SQLModel class User(SQLModel, table=True): id: int = Field(default=None, primary_key=True) name: str email: str ``` 这个类既可以用作 SQLAlchemy 表的映射对象，也可以作为 Pydantic 模型来进行输入/输出验证。 --- #### 3. 编写路由以支持原生查询 当需要执行复杂的原生 SQL 而不是依赖 ORM 方法时，可以通过 SQLAlchemy 的 `execute()` 函数直接发送 SQL 命令给数据库。例如，在 FastAPI 中编写如下路径处理器函数[^1]: ```python from fastapi import Depends, FastAPI from typing import List from sqlmodel import select app = FastAPI() @app.get("/users/", response_model=List[User]) async def read_users(session: AsyncSession = Depends(get_db)): result = await session.execute("SELECT * FROM user") # 原生 SQL 查询 users = result.fetchall() # 将结果转换为 Pydantic 对象列表 pydantic_users = [User.from_orm(user) for user in users] return pydantic_users ``` 在这个示例中，我们通过调用 `await session.execute(...)` 发送一条 SELECT 查询语句到 MySQL 数据库，并获取所有记录的结果集。随后，这些原始元组被逐一传递至 `User.from_orm()` 方法，从而将其转化为对应的 Pydantic 实体形式以便最终返回客户端。 注意这里指定了 `response_model=List[User]` 参数告诉 API 文档该接口预期返回的是多个用户的集合。 --- #### 4. 错误处理机制 如果某些情况下传入的数据不符合预设模式，则 FastAPI 自动触发 HTTPException 返回详细的错误描述信息给前端使用者。比如缺少必填项或者超出长度限制等情况都会被捕获并向外暴露清晰提示消息。 --- ### 总结 综上所述，借助于 SQLModel 可以非常便捷地结合 FastAPI 开发 RESTful Web Services 同时保持高性能表现力的同时还拥有良好的可维护性和扩展性特点。