SQLModel 系统性指南--------------### 目录1. 简介 * 什么是 SQLModel * 为什么使用 SQLModel2. 安装3. 快速入门 * 定义模型 * 创建数据库和表4. 基本 CRUD 操作 * 创建Create * 读取Read * 更新Update * 删除Delete5. 处理关系 * 一对多关系 * 多对多关系6. 高级功能 * 异步支持 * 自定义查询 * 迁移Migrations7. 与 FastAPI 的集成 * 依赖注入 * 路由保护8. 性能优化与最佳实践9. 常见问题解答10. 参考资料* *### 1. 简介#### 什么是 SQLModelSQLModel是一个现代化的 Python 库旨在简化与数据库的交互。它结合了Pydantic和SQLAlchemy的优势使得定义数据模型、进行数据验证和与数据库交互变得更加直观和高效。SQLModel 由Sebastián RamírezFastAPI 的创始人开发专为与 FastAPI 框架无缝集成而设计。#### 为什么使用 SQLModel简洁性通过结合 Pydantic 的数据验证和 SQLAlchemy 的 ORM 功能SQLModel 使模型定义和数据库操作更加简洁。*类型安全充分利用 Python 的类型提示增强代码的可读性和可靠性。*与 FastAPI 无缝集成优化了与 FastAPI 的集成支持自动文档生成和依赖注入。*灵活性支持同步和异步操作适应不同的性能需求。*现代化设计采用现代化的 Python 编码风格和最佳实践提升开发体验。* *### 2. 安装首先确保您已经安装了 Python 3.7 或更高版本。然后使用pip安装sqlmodel包 pip install sqlmodel 此外根据您使用的数据库还需要安装相应的数据库驱动。例如SQLite无需额外安装驱动Python 内置支持。 *PostgreSQL pip install asyncpg *MySQL pip install pymysql * *### 3. 快速入门#### 定义模型使用SQLModel定义数据模型时通常会继承自SQLModel并使用tableTrue参数指示这是一个数据库表。 from typing import Optional from sqlmodel import SQLModel, Field from datetime import datetime class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: str Field(indexTrue, nullableFalse, uniqueTrue) hashed_password: str Field(nullableFalse) is_active: bool Field(defaultTrue) created_at: datetime Field(default_factorydatetime.utcnow) #### 创建数据库和表使用 SQLAlchemy 的引擎和SQLModel的元数据来创建数据库和表。 from sqlmodel import SQLModel, create_engine from models import User # 假设上面的模型保存在 models.py 文件中 DATABASE_URL “sqlite:///./test.db” # 或者使用其他数据库如 PostgreSQL engine create_engine(DATABASE_URL, echoTrue) def create_db_and_tables(): SQLModel.metadata.create_all(engine) 在应用启动时调用create_db_and_tables来创建数据库表。*### 4. 基本 CRUD 操作#### 创建Create向数据库中插入一条新记录。 from sqlmodel import Session, select from models import User from database import engine, create_db_and_tables def create_user(username: str, email: str, hashed_password: str) - User: user User(usernameusername, emailemail, hashed_passwordhashed_password) with Session(engine) as session: session.add(user) session.commit() session.refresh(user) return user #### 读取Read从数据库中查询记录。 def get_user_by_id(user_id: int) - Optional[User]: with Session(engine) as session: user session.get(User, user_id) return user def get_user_by_username(username: str) - Optional[User]: with Session(engine) as session: statement select(User).where(User.username username) user session.exec(statement).first() return user #### 更新Update更新数据库中的记录。 def update_user_email(user_id: int, new_email: str) - Optional[User]: with Session(engine) as session: user session.get(User, user_id) if user: user.email new_email session.add(user) session.commit() session.refresh(user) return user return None #### 删除Delete从数据库中删除记录。 def delete_user(user_id: int) - bool: with Session(engine) as session: user session.get(User, user_id) if user: session.delete(user) session.commit() return True return False * *### 5. 处理关系#### 一对多关系例如一个用户可以有多条地址记录。 from typing import List, Optional from sqlmodel import SQLModel, Field, Relationship class Address(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) street: str city: str user_id: int Field(foreign_key“user.id”) user: Optional[“User”] Relationship(back_populates“addresses”) class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: str Field(indexTrue, nullableFalse, uniqueTrue) hashed_password: str Field(nullableFalse) is_active: bool Field(defaultTrue) created_at: datetime Field(default_factorydatetime.utcnow) addresses: List[Address] Relationship(back_populates“user”) #### 多对多关系例如用户和角色之间的多对多关系。 from typing import List, Optional from sqlmodel import SQLModel, Field, Relationship class UserRoleLink(SQLModel, tableTrue): user_id: int Field(foreign_key“user.id”, primary_keyTrue) role_id: int Field(foreign_key“role.id”, primary_keyTrue) class Role(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) name: str users: List[“User”] Relationship( back_populates“roles”, link_modelUserRoleLink ) class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: str Field(indexTrue, nullableFalse, uniqueTrue) hashed_password: str Field(nullableFalse) is_active: bool Field(defaultTrue) created_at: datetime Field(default_factorydatetime.utcnow) roles: List[Role] Relationship( back_populates“users”, link_modelUserRoleLink ) * *### 6. 高级功能#### 异步支持SQLModel支持异步数据库操作适用于需要高并发和高性能的应用。首先安装异步驱动如asyncpg用于 PostgreSQL pip install asyncpg 然后配置异步引擎和会话 from sqlmodel import SQLModel, create_engine, select from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker from models import User from datetime import datetime DATABASE_URL “postgresqlasyncpg://user:passwordlocalhost/dbname” async_engine create_async_engine(DATABASE_URL, echoTrue) async_session sessionmaker( async_engine, class_AsyncSession, expire_on_commitFalse ) async def init_db(): async with async_engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all) # 在应用启动时调用 init_db import asyncio asyncio.run(init_db()) # 异步获取会话 async def get_async_session(): async with async_session() as session: yield session # 异步 CRUD 操作示例 async def get_user_async(user_id: int) - Optional[User]: async with async_session() as session: user await session.get(User, user_id) return user #### 自定义查询使用 SQLAlchemy 的强大查询功能执行复杂的数据库操作。 from sqlmodel import Session, select, func from models import User def count_users() - int: with Session(engine) as session: statement select(func.count(User.id)) count session.exec(statement).one() return count def get_users_with_email_domain(domain: str) - List[User]: with Session(engine) as session: statement select(User).where(User.email.like(f%{domain})) users session.exec(statement).all() return users #### 迁移Migrations虽然SQLModel本身不提供迁移工具但它与 Alembic 完全兼容可以使用 Alembic 进行数据库迁移。安装 Alembic pip install alembic初始化 Alembic alembic init alembic配置 Alembic编辑alembic.ini设置sqlalchemy.url为您的数据库 URL。在alembic/env.py中导入您的模型 from logging.config import fileConfig from sqlalchemy import engine_from_config from sqlalchemy import pool from sqlmodel import SQLModel import sys import os # 将项目路径添加到 sys.path sys.path.append(os.path.dirname(os.path.dirname(file))) from models import User # 导入您的模型 # this is the Alembic Config object, which provides # access to the values within the .ini file in use. config context.config # Interpret the config file for Python logging. # This line sets up loggers basically. fileConfig(config.config_file_name) target_metadata SQLModel.metadata def run_migrations_offline(): … # 保持默认配置 def run_migrations_online(): … # 保持默认配置 if context.is_offline_mode(): run_migrations_offline() else: run_migrations_online()创建迁移脚本 alembic revision --autogenerate -m “Initial migration”应用迁移 alembic upgrade head * *### 7. 与 FastAPI 的集成#### 依赖注入利用 FastAPI 的依赖注入机制将数据库会话注入到路由中。 from fastapi import FastAPI, Depends, HTTPException from sqlmodel import Session, select from models import User from database import engine, get_session app FastAPI() app.post(“/users/”, response_modelUser) def create_user(user: User, session: Session Depends(get_session)): db_user session.exec(select(User).where(User.username user.username)).first() if db_user: raise HTTPException(status_code400, detail“Username already exists”) session.add(user) session.commit() session.refresh(user) return user app.get(“/users/{user_id}”, response_modelUser) def read_user(user_id: int, session: Session Depends(get_session)): user session.get(User, user_id) if not user: raise HTTPException(status_code404, detail“User not found”) return user #### 路由保护结合 JWT 进行身份验证保护特定路由。安装fastapi-jwt-auth pip install fastapi-jwt-auth配置 JWT from fastapi import FastAPI, Depends, HTTPException from fastapi_jwt_auth import AuthJWT from pydantic import BaseModel from sqlmodel import Session, select from models import User from database import engine, get_session class Settings(BaseModel): authjwt_secret_key: str “your-secret-key” app FastAPI() AuthJWT.load_config def get_config(): return Settings() app.post(‘/login’) def login(user: User, Authorize: AuthJWT Depends()): # 验证用户凭证此处省略具体验证逻辑 access_token Authorize.create_access_token(subjectuser.username) return {“access_token”: access_token} app.get(‘/protected’) def protected(Authorize: AuthJWT Depends()): Authorize.jwt_required() current_user Authorize.get_jwt_subject() return {“message”: fHello, {current_user}} * *### 8. 性能优化与最佳实践#### 8.1 使用连接池优化数据库连接使用连接池以提高性能和资源利用率。 from sqlmodel import create_engine DATABASE_URL “sqlite:///./test.db” engine create_engine(DATABASE_URL, echoTrue, pool_size20, max_overflow0) #### 8.2 异步操作对于高并发应用使用异步数据库操作。 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker DATABASE_URL “postgresqlasyncpg://user:passwordlocalhost/dbname” async_engine create_async_engine(DATABASE_URL, echoTrue) async_session sessionmaker( async_engine, class_AsyncSession, expire_on_commitFalse ) #### 8.3 缓存使用缓存机制如 Redis减少数据库查询提高响应速度。 import redis redis_client redis.Redis(host‘localhost’, port6379, db0) def get_user_cached(user_id: int) - Optional[User]: cached_user redis_client.get(fuser:{user_id}“) if cached_user: return User.parse_raw(cached_user) with Session(engine) as session: user session.get(User, user_id) if user: redis_client.set(fuser:{user_id}”, user.json(), ex3600) return user #### 8.4 索引优化为常用查询字段添加索引提高查询性能。 class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: str Field(indexTrue, nullableFalse, uniqueTrue) # 其他字段… #### 8.5 分页对于大量数据查询使用分页机制减少单次查询的数据量。 def get_users_paginated(skip: int 0, limit: int 10) - List[User]: with Session(engine) as session: statement select(User).offset(skip).limit(limit) users session.exec(statement).all() return users * *### 9. 常见问题解答#### 9.1 如何在 SQLModel 中使用外键在定义模型时使用Field的foreign_key参数指定外键。 class Address(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) street: str city: str user_id: int Field(foreign_key“user.id”) user: Optional[“User”] Relationship(back_populates“addresses”) #### 9.2 SQLModel 支持哪些数据库SQLModel基于 SQLAlchemy支持所有 SQLAlchemy 支持的数据库包括SQLitePostgreSQLMySQLSQL ServerOracle以及其他数据库通过相应的数据库驱动支持。#### 9.3 如何进行数据库迁移SQLModel本身不提供迁移工具但可以与Alembic配合使用进行数据库迁移。安装 Alembic pip install alembic初始化 Alembic alembic init alembic配置 Alembic编辑alembic.ini设置sqlalchemy.url为您的数据库 URL。在alembic/env.py中导入您的模型 from logging.config import fileConfig from sqlalchemy import engine_from_config from sqlalchemy import pool from sqlmodel import SQLModel import sys import os # 将项目路径添加到 sys.path sys.path.append(os.path.dirname(os.path.dirname(file))) from models import User # 导入您的模型 config context.config fileConfig(config.config_file_name) target_metadata SQLModel.metadata def run_migrations_offline(): … def run_migrations_online(): … if context.is_offline_mode(): run_migrations_offline() else: run_migrations_online()创建迁移脚本 alembic revision --autogenerate -m “Initial migration”应用迁移 alembic upgrade head #### 9.4 如何处理模型验证错误SQLModel结合了 Pydantic 的数据验证功能可以在模型定义中使用 Pydantic 的字段验证器。 from sqlmodel import SQLModel, Field from pydantic import validator, EmailStr class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: EmailStr Field(indexTrue, nullableFalse, uniqueTrue) hashed_password: str Field(nullableFalse) validator(‘username’) def username_must_not_be_empty(cls, v): if not v or not v.strip(): raise ValueError(‘Username must not be empty’) return v * *### 10. 参考资料SQLModel 官方文档https://sqlmodel.tiangolo.com/*SQLAlchemy 官方文档https://www.sqlalchemy.org/*FastAPI 官方文档https://fastapi.tiangolo.com/*Alembic 官方文档https://alembic.sqlalchemy.org/en/latest/*Real Python 的 SQLModel 教程https://realpython.com/sqlmodel-python-orm/*Pydantic 官方文档https://pydantic-docs.helpmanual.io/*GitHub 上的 SQLModel 仓库https://github.com/tiangolo/sqlmodel* * *### 附录完整示例以下是一个完整的 FastAPI 应用示例展示了如何使用 SQLModel 进行数据库操作和 API 构建。#### 目录结构 my_fastapi_app/ ├── main.py ├── models.py ├── database.py ├── schemas.py └── alembic/ ├── env.py ├── script.py.mako └── versions/ ####models.pyfrom typing import List, Optional from sqlmodel import SQLModel, Field, Relationship from datetime import datetime class Address(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) street: str city: str user_id: int Field(foreign_key“user.id”) user: Optional[“User”] Relationship(back_populates“addresses”) class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: str Field(indexTrue, nullableFalse, uniqueTrue) hashed_password: str Field(nullableFalse) is_active: bool Field(defaultTrue) created_at: datetime Field(default_factorydatetime.utcnow) addresses: List[Address] Relationship(back_populates“user”) ####schemas.pyfrom typing import List, Optional from pydantic import BaseModel, EmailStr from datetime import datetime class AddressCreate(BaseModel): street: str city: str class AddressRead(BaseModel): id: int street: str city: str class Config: orm_mode True class UserCreate(BaseModel): username: str email: EmailStr password: str class UserRead(BaseModel): id: int username: str email: EmailStr is_active: bool created_at: datetime addresses: List[AddressRead] [] class Config: orm_mode True ####database.pyfrom sqlmodel import SQLModel, create_engine, Session from models import User, Address DATABASE_URL “sqlite:///./test.db” engine create_engine(DATABASE_URL, echoTrue) def create_db_and_tables(): SQLModel.metadata.create_all(engine) def get_session(): with Session(engine) as session: yield session ####main.pyfrom fastapi import FastAPI, Depends, HTTPException from sqlmodel import Session, select from models import User, Address from schemas import UserCreate, UserRead, AddressCreate, AddressRead from database import create_db_and_tables, get_session from typing import List app FastAPI() app.on_event(“startup”) def on_startup(): create_db_and_tables() app.post(“/users/”, response_modelUserRead) def create_user(user: UserCreate, session: Session Depends(get_session)): db_user session.exec(select(User).where(User.username user.username)).first() if db_user: raise HTTPException(status_code400, detail“Username already exists”) new_user User( usernameuser.username, emailuser.email, hashed_passworduser.password # 实际项目中应进行哈希处理 ) session.add(new_user) session.commit() session.refresh(new_user) return new_user app.get(“/users/{user_id}”, response_modelUserRead) def read_user(user_id: int, session: Session Depends(get_session)): user session.get(User, user_id) if not user: raise HTTPException(status_code404, detail“User not found”) return user app.post(“/users/{user_id}/addresses/”, response_modelAddressRead) def create_address(user_id: int, address: AddressCreate, session: Session Depends(get_session)): user session.get(User, user_id) if not user: raise HTTPException(status_code404, detail“User not found”) new_address Address(*address.dict(), user_iduser_id) session.add(new_address) session.commit() session.refresh(new_address) return new_address app.get(“/users/{user_id}/addresses/”, response_modelList[AddressRead]) def read_addresses(user_id: int, session: Session Depends(get_session)): user session.get(User, user_id) if not user: raise HTTPException(status_code404, detail“User not found”) return user.addresses #### 运行应用使用uvicorn运行 FastAPI 应用 uvicorn main:app --reload 访问 http://127.0.0.1:8000/docs 查看自动生成的 API 文档并进行测试。* * SQLModel入门
SQLModel 系统性指南--------------### 目录1. 简介 * 什么是 SQLModel * 为什么使用 SQLModel2. 安装3. 快速入门 * 定义模型 * 创建数据库和表4. 基本 CRUD 操作 * 创建Create * 读取Read * 更新Update * 删除Delete5. 处理关系 * 一对多关系 * 多对多关系6. 高级功能 * 异步支持 * 自定义查询 * 迁移Migrations7. 与 FastAPI 的集成 * 依赖注入 * 路由保护8. 性能优化与最佳实践9. 常见问题解答10. 参考资料* *### 1. 简介#### 什么是 SQLModelSQLModel是一个现代化的 Python 库旨在简化与数据库的交互。它结合了Pydantic和SQLAlchemy的优势使得定义数据模型、进行数据验证和与数据库交互变得更加直观和高效。SQLModel 由Sebastián RamírezFastAPI 的创始人开发专为与 FastAPI 框架无缝集成而设计。#### 为什么使用 SQLModel简洁性通过结合 Pydantic 的数据验证和 SQLAlchemy 的 ORM 功能SQLModel 使模型定义和数据库操作更加简洁。*类型安全充分利用 Python 的类型提示增强代码的可读性和可靠性。*与 FastAPI 无缝集成优化了与 FastAPI 的集成支持自动文档生成和依赖注入。*灵活性支持同步和异步操作适应不同的性能需求。*现代化设计采用现代化的 Python 编码风格和最佳实践提升开发体验。* *### 2. 安装首先确保您已经安装了 Python 3.7 或更高版本。然后使用pip安装sqlmodel包 pip install sqlmodel 此外根据您使用的数据库还需要安装相应的数据库驱动。例如SQLite无需额外安装驱动Python 内置支持。 *PostgreSQL pip install asyncpg *MySQL pip install pymysql * *### 3. 快速入门#### 定义模型使用SQLModel定义数据模型时通常会继承自SQLModel并使用tableTrue参数指示这是一个数据库表。 from typing import Optional from sqlmodel import SQLModel, Field from datetime import datetime class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: str Field(indexTrue, nullableFalse, uniqueTrue) hashed_password: str Field(nullableFalse) is_active: bool Field(defaultTrue) created_at: datetime Field(default_factorydatetime.utcnow) #### 创建数据库和表使用 SQLAlchemy 的引擎和SQLModel的元数据来创建数据库和表。 from sqlmodel import SQLModel, create_engine from models import User # 假设上面的模型保存在 models.py 文件中 DATABASE_URL “sqlite:///./test.db” # 或者使用其他数据库如 PostgreSQL engine create_engine(DATABASE_URL, echoTrue) def create_db_and_tables(): SQLModel.metadata.create_all(engine) 在应用启动时调用create_db_and_tables来创建数据库表。*### 4. 基本 CRUD 操作#### 创建Create向数据库中插入一条新记录。 from sqlmodel import Session, select from models import User from database import engine, create_db_and_tables def create_user(username: str, email: str, hashed_password: str) - User: user User(usernameusername, emailemail, hashed_passwordhashed_password) with Session(engine) as session: session.add(user) session.commit() session.refresh(user) return user #### 读取Read从数据库中查询记录。 def get_user_by_id(user_id: int) - Optional[User]: with Session(engine) as session: user session.get(User, user_id) return user def get_user_by_username(username: str) - Optional[User]: with Session(engine) as session: statement select(User).where(User.username username) user session.exec(statement).first() return user #### 更新Update更新数据库中的记录。 def update_user_email(user_id: int, new_email: str) - Optional[User]: with Session(engine) as session: user session.get(User, user_id) if user: user.email new_email session.add(user) session.commit() session.refresh(user) return user return None #### 删除Delete从数据库中删除记录。 def delete_user(user_id: int) - bool: with Session(engine) as session: user session.get(User, user_id) if user: session.delete(user) session.commit() return True return False * *### 5. 处理关系#### 一对多关系例如一个用户可以有多条地址记录。 from typing import List, Optional from sqlmodel import SQLModel, Field, Relationship class Address(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) street: str city: str user_id: int Field(foreign_key“user.id”) user: Optional[“User”] Relationship(back_populates“addresses”) class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: str Field(indexTrue, nullableFalse, uniqueTrue) hashed_password: str Field(nullableFalse) is_active: bool Field(defaultTrue) created_at: datetime Field(default_factorydatetime.utcnow) addresses: List[Address] Relationship(back_populates“user”) #### 多对多关系例如用户和角色之间的多对多关系。 from typing import List, Optional from sqlmodel import SQLModel, Field, Relationship class UserRoleLink(SQLModel, tableTrue): user_id: int Field(foreign_key“user.id”, primary_keyTrue) role_id: int Field(foreign_key“role.id”, primary_keyTrue) class Role(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) name: str users: List[“User”] Relationship( back_populates“roles”, link_modelUserRoleLink ) class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: str Field(indexTrue, nullableFalse, uniqueTrue) hashed_password: str Field(nullableFalse) is_active: bool Field(defaultTrue) created_at: datetime Field(default_factorydatetime.utcnow) roles: List[Role] Relationship( back_populates“users”, link_modelUserRoleLink ) * *### 6. 高级功能#### 异步支持SQLModel支持异步数据库操作适用于需要高并发和高性能的应用。首先安装异步驱动如asyncpg用于 PostgreSQL pip install asyncpg 然后配置异步引擎和会话 from sqlmodel import SQLModel, create_engine, select from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker from models import User from datetime import datetime DATABASE_URL “postgresqlasyncpg://user:passwordlocalhost/dbname” async_engine create_async_engine(DATABASE_URL, echoTrue) async_session sessionmaker( async_engine, class_AsyncSession, expire_on_commitFalse ) async def init_db(): async with async_engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all) # 在应用启动时调用 init_db import asyncio asyncio.run(init_db()) # 异步获取会话 async def get_async_session(): async with async_session() as session: yield session # 异步 CRUD 操作示例 async def get_user_async(user_id: int) - Optional[User]: async with async_session() as session: user await session.get(User, user_id) return user #### 自定义查询使用 SQLAlchemy 的强大查询功能执行复杂的数据库操作。 from sqlmodel import Session, select, func from models import User def count_users() - int: with Session(engine) as session: statement select(func.count(User.id)) count session.exec(statement).one() return count def get_users_with_email_domain(domain: str) - List[User]: with Session(engine) as session: statement select(User).where(User.email.like(f%{domain})) users session.exec(statement).all() return users #### 迁移Migrations虽然SQLModel本身不提供迁移工具但它与 Alembic 完全兼容可以使用 Alembic 进行数据库迁移。安装 Alembic pip install alembic初始化 Alembic alembic init alembic配置 Alembic编辑alembic.ini设置sqlalchemy.url为您的数据库 URL。在alembic/env.py中导入您的模型 from logging.config import fileConfig from sqlalchemy import engine_from_config from sqlalchemy import pool from sqlmodel import SQLModel import sys import os # 将项目路径添加到 sys.path sys.path.append(os.path.dirname(os.path.dirname(file))) from models import User # 导入您的模型 # this is the Alembic Config object, which provides # access to the values within the .ini file in use. config context.config # Interpret the config file for Python logging. # This line sets up loggers basically. fileConfig(config.config_file_name) target_metadata SQLModel.metadata def run_migrations_offline(): … # 保持默认配置 def run_migrations_online(): … # 保持默认配置 if context.is_offline_mode(): run_migrations_offline() else: run_migrations_online()创建迁移脚本 alembic revision --autogenerate -m “Initial migration”应用迁移 alembic upgrade head * *### 7. 与 FastAPI 的集成#### 依赖注入利用 FastAPI 的依赖注入机制将数据库会话注入到路由中。 from fastapi import FastAPI, Depends, HTTPException from sqlmodel import Session, select from models import User from database import engine, get_session app FastAPI() app.post(“/users/”, response_modelUser) def create_user(user: User, session: Session Depends(get_session)): db_user session.exec(select(User).where(User.username user.username)).first() if db_user: raise HTTPException(status_code400, detail“Username already exists”) session.add(user) session.commit() session.refresh(user) return user app.get(“/users/{user_id}”, response_modelUser) def read_user(user_id: int, session: Session Depends(get_session)): user session.get(User, user_id) if not user: raise HTTPException(status_code404, detail“User not found”) return user #### 路由保护结合 JWT 进行身份验证保护特定路由。安装fastapi-jwt-auth pip install fastapi-jwt-auth配置 JWT from fastapi import FastAPI, Depends, HTTPException from fastapi_jwt_auth import AuthJWT from pydantic import BaseModel from sqlmodel import Session, select from models import User from database import engine, get_session class Settings(BaseModel): authjwt_secret_key: str “your-secret-key” app FastAPI() AuthJWT.load_config def get_config(): return Settings() app.post(‘/login’) def login(user: User, Authorize: AuthJWT Depends()): # 验证用户凭证此处省略具体验证逻辑 access_token Authorize.create_access_token(subjectuser.username) return {“access_token”: access_token} app.get(‘/protected’) def protected(Authorize: AuthJWT Depends()): Authorize.jwt_required() current_user Authorize.get_jwt_subject() return {“message”: fHello, {current_user}} * *### 8. 性能优化与最佳实践#### 8.1 使用连接池优化数据库连接使用连接池以提高性能和资源利用率。 from sqlmodel import create_engine DATABASE_URL “sqlite:///./test.db” engine create_engine(DATABASE_URL, echoTrue, pool_size20, max_overflow0) #### 8.2 异步操作对于高并发应用使用异步数据库操作。 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker DATABASE_URL “postgresqlasyncpg://user:passwordlocalhost/dbname” async_engine create_async_engine(DATABASE_URL, echoTrue) async_session sessionmaker( async_engine, class_AsyncSession, expire_on_commitFalse ) #### 8.3 缓存使用缓存机制如 Redis减少数据库查询提高响应速度。 import redis redis_client redis.Redis(host‘localhost’, port6379, db0) def get_user_cached(user_id: int) - Optional[User]: cached_user redis_client.get(fuser:{user_id}“) if cached_user: return User.parse_raw(cached_user) with Session(engine) as session: user session.get(User, user_id) if user: redis_client.set(fuser:{user_id}”, user.json(), ex3600) return user #### 8.4 索引优化为常用查询字段添加索引提高查询性能。 class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: str Field(indexTrue, nullableFalse, uniqueTrue) # 其他字段… #### 8.5 分页对于大量数据查询使用分页机制减少单次查询的数据量。 def get_users_paginated(skip: int 0, limit: int 10) - List[User]: with Session(engine) as session: statement select(User).offset(skip).limit(limit) users session.exec(statement).all() return users * *### 9. 常见问题解答#### 9.1 如何在 SQLModel 中使用外键在定义模型时使用Field的foreign_key参数指定外键。 class Address(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) street: str city: str user_id: int Field(foreign_key“user.id”) user: Optional[“User”] Relationship(back_populates“addresses”) #### 9.2 SQLModel 支持哪些数据库SQLModel基于 SQLAlchemy支持所有 SQLAlchemy 支持的数据库包括SQLitePostgreSQLMySQLSQL ServerOracle以及其他数据库通过相应的数据库驱动支持。#### 9.3 如何进行数据库迁移SQLModel本身不提供迁移工具但可以与Alembic配合使用进行数据库迁移。安装 Alembic pip install alembic初始化 Alembic alembic init alembic配置 Alembic编辑alembic.ini设置sqlalchemy.url为您的数据库 URL。在alembic/env.py中导入您的模型 from logging.config import fileConfig from sqlalchemy import engine_from_config from sqlalchemy import pool from sqlmodel import SQLModel import sys import os # 将项目路径添加到 sys.path sys.path.append(os.path.dirname(os.path.dirname(file))) from models import User # 导入您的模型 config context.config fileConfig(config.config_file_name) target_metadata SQLModel.metadata def run_migrations_offline(): … def run_migrations_online(): … if context.is_offline_mode(): run_migrations_offline() else: run_migrations_online()创建迁移脚本 alembic revision --autogenerate -m “Initial migration”应用迁移 alembic upgrade head #### 9.4 如何处理模型验证错误SQLModel结合了 Pydantic 的数据验证功能可以在模型定义中使用 Pydantic 的字段验证器。 from sqlmodel import SQLModel, Field from pydantic import validator, EmailStr class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: EmailStr Field(indexTrue, nullableFalse, uniqueTrue) hashed_password: str Field(nullableFalse) validator(‘username’) def username_must_not_be_empty(cls, v): if not v or not v.strip(): raise ValueError(‘Username must not be empty’) return v * *### 10. 参考资料SQLModel 官方文档https://sqlmodel.tiangolo.com/*SQLAlchemy 官方文档https://www.sqlalchemy.org/*FastAPI 官方文档https://fastapi.tiangolo.com/*Alembic 官方文档https://alembic.sqlalchemy.org/en/latest/*Real Python 的 SQLModel 教程https://realpython.com/sqlmodel-python-orm/*Pydantic 官方文档https://pydantic-docs.helpmanual.io/*GitHub 上的 SQLModel 仓库https://github.com/tiangolo/sqlmodel* * *### 附录完整示例以下是一个完整的 FastAPI 应用示例展示了如何使用 SQLModel 进行数据库操作和 API 构建。#### 目录结构 my_fastapi_app/ ├── main.py ├── models.py ├── database.py ├── schemas.py └── alembic/ ├── env.py ├── script.py.mako └── versions/ ####models.pyfrom typing import List, Optional from sqlmodel import SQLModel, Field, Relationship from datetime import datetime class Address(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) street: str city: str user_id: int Field(foreign_key“user.id”) user: Optional[“User”] Relationship(back_populates“addresses”) class User(SQLModel, tableTrue): id: Optional[int] Field(defaultNone, primary_keyTrue) username: str Field(indexTrue, nullableFalse, uniqueTrue) email: str Field(indexTrue, nullableFalse, uniqueTrue) hashed_password: str Field(nullableFalse) is_active: bool Field(defaultTrue) created_at: datetime Field(default_factorydatetime.utcnow) addresses: List[Address] Relationship(back_populates“user”) ####schemas.pyfrom typing import List, Optional from pydantic import BaseModel, EmailStr from datetime import datetime class AddressCreate(BaseModel): street: str city: str class AddressRead(BaseModel): id: int street: str city: str class Config: orm_mode True class UserCreate(BaseModel): username: str email: EmailStr password: str class UserRead(BaseModel): id: int username: str email: EmailStr is_active: bool created_at: datetime addresses: List[AddressRead] [] class Config: orm_mode True ####database.pyfrom sqlmodel import SQLModel, create_engine, Session from models import User, Address DATABASE_URL “sqlite:///./test.db” engine create_engine(DATABASE_URL, echoTrue) def create_db_and_tables(): SQLModel.metadata.create_all(engine) def get_session(): with Session(engine) as session: yield session ####main.pyfrom fastapi import FastAPI, Depends, HTTPException from sqlmodel import Session, select from models import User, Address from schemas import UserCreate, UserRead, AddressCreate, AddressRead from database import create_db_and_tables, get_session from typing import List app FastAPI() app.on_event(“startup”) def on_startup(): create_db_and_tables() app.post(“/users/”, response_modelUserRead) def create_user(user: UserCreate, session: Session Depends(get_session)): db_user session.exec(select(User).where(User.username user.username)).first() if db_user: raise HTTPException(status_code400, detail“Username already exists”) new_user User( usernameuser.username, emailuser.email, hashed_passworduser.password # 实际项目中应进行哈希处理 ) session.add(new_user) session.commit() session.refresh(new_user) return new_user app.get(“/users/{user_id}”, response_modelUserRead) def read_user(user_id: int, session: Session Depends(get_session)): user session.get(User, user_id) if not user: raise HTTPException(status_code404, detail“User not found”) return user app.post(“/users/{user_id}/addresses/”, response_modelAddressRead) def create_address(user_id: int, address: AddressCreate, session: Session Depends(get_session)): user session.get(User, user_id) if not user: raise HTTPException(status_code404, detail“User not found”) new_address Address(*address.dict(), user_iduser_id) session.add(new_address) session.commit() session.refresh(new_address) return new_address app.get(“/users/{user_id}/addresses/”, response_modelList[AddressRead]) def read_addresses(user_id: int, session: Session Depends(get_session)): user session.get(User, user_id) if not user: raise HTTPException(status_code404, detail“User not found”) return user.addresses #### 运行应用使用uvicorn运行 FastAPI 应用 uvicorn main:app --reload 访问 http://127.0.0.1:8000/docs 查看自动生成的 API 文档并进行测试。* *