sqlmodel

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

SQLModel Development Guide

SQLModel 开发指南

SQLModel combines SQLAlchemy and Pydantic into a single library - one model class serves as both ORM model and Pydantic schema.

SQLModel 将 SQLAlchemy 和 Pydantic 整合到一个库中——一个模型类可同时作为 ORM 模型和 Pydantic 模式使用。

Quick Start

快速开始

Installation

安装

bash

pip install sqlmodel

bash

pip install sqlmodel

Minimal Example

最简示例

python

from sqlmodel import Field, SQLModel, Session, create_engine, select

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    age: int | None = None

engine = create_engine("sqlite:///database.db")
SQLModel.metadata.create_all(engine)

python

from sqlmodel import Field, SQLModel, Session, create_engine, select

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    age: int | None = None

engine = create_engine("sqlite:///database.db")
SQLModel.metadata.create_all(engine)

Create

创建

with Session(engine) as session: hero = Hero(name="Spider-Boy", age=18) session.add(hero) session.commit() session.refresh(hero)

Read

读取

with Session(engine) as session: heroes = session.exec(select(Hero)).all()

undefined

with Session(engine) as session: heroes = session.exec(select(Hero)).all()

undefined

Core Concepts

核心概念

Concept	Description
`table=True`	Makes class a database table (without it, it's just Pydantic)
`Field()`	Define column attributes: `primary_key` , `index` , `unique` , `foreign_key`
`Session`	Database session for CRUD operations
`select()`	Type-safe query builder
`Relationship`	Define relationships between models

概念	描述
`table=True`	将类设为数据库表（不添加该参数时，仅作为 Pydantic 模式）
`Field()`	定义列属性： `primary_key` 、 `index` 、 `unique` 、 `foreign_key`
`Session`	用于 CRUD 操作的数据库会话
`select()`	类型安全的查询构建器
`Relationship`	定义模型之间的关系

Model Patterns

模型模式

Base Model (API Schema Only)

基础模型（仅 API 模式）

python

class HeroBase(SQLModel):
    name: str
    age: int | None = None

python

class HeroBase(SQLModel):
    name: str
    age: int | None = None

Table Model (Database)

表模型（数据库）

python

class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)

python

class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)

Request/Response Models

请求/响应模型

python

class HeroCreate(HeroBase):
    secret_name: str

class HeroPublic(HeroBase):
    id: int

class HeroUpdate(SQLModel):
    name: str | None = None
    age: int | None = None

python

class HeroCreate(HeroBase):
    secret_name: str

class HeroPublic(HeroBase):
    id: int

class HeroUpdate(SQLModel):
    name: str | None = None
    age: int | None = None

CRUD Operations

CRUD 操作

Create

创建

python

def create_hero(session: Session, hero: HeroCreate) -> Hero:
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

python

def create_hero(session: Session, hero: HeroCreate) -> Hero:
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

Read

读取

python

def get_hero(session: Session, hero_id: int) -> Hero | None:
    return session.get(Hero, hero_id)

def get_heroes(session: Session, skip: int = 0, limit: int = 100) -> list[Hero]:
    return session.exec(select(Hero).offset(skip).limit(limit)).all()

python

def get_hero(session: Session, hero_id: int) -> Hero | None:
    return session.get(Hero, hero_id)

def get_heroes(session: Session, skip: int = 0, limit: int = 100) -> list[Hero]:
    return session.exec(select(Hero).offset(skip).limit(limit)).all()

Update

更新

python

def update_hero(session: Session, hero_id: int, hero_update: HeroUpdate) -> Hero | None:
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        return None
    hero_data = hero_update.model_dump(exclude_unset=True)
    db_hero.sqlmodel_update(hero_data)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

python

def update_hero(session: Session, hero_id: int, hero_update: HeroUpdate) -> Hero | None:
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        return None
    hero_data = hero_update.model_dump(exclude_unset=True)
    db_hero.sqlmodel_update(hero_data)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

Delete

删除

python

def delete_hero(session: Session, hero_id: int) -> bool:
    hero = session.get(Hero, hero_id)
    if not hero:
        return False
    session.delete(hero)
    session.commit()
    return True

python

def delete_hero(session: Session, hero_id: int) -> bool:
    hero = session.get(Hero, hero_id)
    if not hero:
        return False
    session.delete(hero)
    session.commit()
    return True

FastAPI Integration

FastAPI 集成

Database Setup

数据库设置

python

from sqlmodel import SQLModel, Session, create_engine

DATABASE_URL = "sqlite:///./database.db"
engine = create_engine(DATABASE_URL, echo=True)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

python

from sqlmodel import SQLModel, Session, create_engine

DATABASE_URL = "sqlite:///./database.db"
engine = create_engine(DATABASE_URL, echo=True)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

Dependency Injection

依赖注入

python

from typing import Annotated
from fastapi import Depends

SessionDep = Annotated[Session, Depends(get_session)]

@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

python

from typing import Annotated
from fastapi import Depends

SessionDep = Annotated[Session, Depends(get_session)]

@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

Lifespan Events

生命周期事件

python

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)

python

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)

Reference Files

参考文档

Load these based on the task at hand:

Topic	File	When to Use
Models	models.md	Field options, validators, computed fields, inheritance, mixins
Relationships	relationships.md	One-to-many, many-to-many, self-referential, lazy loading
Async	async.md	Async sessions, async engine, background tasks
Migrations	migrations.md	Alembic setup, auto-generation, migration patterns

根据任务需求加载以下文档：

主题	文件	使用场景
模型	models.md	字段选项、验证器、计算字段、继承、混合类
关系	relationships.md	一对多、多对多、自引用、延迟加载
异步	async.md	异步会话、异步引擎、后台任务
迁移	migrations.md	Alembic 配置、自动生成、迁移模式

Querying

查询

Basic Queries

基础查询

python

undefined

python

undefined

All heroes

所有英雄

heroes = session.exec(select(Hero)).all()

Single result (first or None)

单个结果（第一个或空）

hero = session.exec(select(Hero).where(Hero.name == "Spider-Boy")).first()

Get by primary key

通过主键获取

hero = session.get(Hero, 1)

undefined

hero = session.get(Hero, 1)

undefined

Filtering

过滤

python

from sqlmodel import select, or_, and_

python

from sqlmodel import select, or_, and_

Single condition

单个条件

select(Hero).where(Hero.age >= 18)

Multiple conditions (AND)

多个条件（AND）

select(Hero).where(Hero.age >= 18, Hero.name == "Spider-Boy")

OR conditions

OR 条件

select(Hero).where(or_(Hero.age < 18, Hero.age > 60))

LIKE/contains

LIKE/包含

select(Hero).where(Hero.name.contains("Spider"))

undefined

select(Hero).where(Hero.name.contains("Spider"))

undefined

Ordering and Pagination

排序和分页

python

select(Hero).order_by(Hero.name)
select(Hero).order_by(Hero.age.desc())
select(Hero).offset(10).limit(5)

python

select(Hero).order_by(Hero.name)
select(Hero).order_by(Hero.age.desc())
select(Hero).offset(10).limit(5)

Best Practices

最佳实践

Separate table models from API schemas - Use
```
table=True
```
only for actual DB tables
Use
model_validate()
for conversion - Convert between schemas and table models

Use
sqlmodel_update()
for partial updates - Pass

exclude_unset=True

model_dump()

Always use
Field()
for constraints - Primary keys, indexes, foreign keys, defaults
Use
Annotated
dependencies - Clean, reusable session injection
Use lifespan for table creation - Not deprecated
```
@app.on_event
```
Index frequently queried columns -
```
Field(index=True)
```
Use
echo=True
during development - See generated SQL queries

分离表模型与 API 模式 - 仅为实际数据库表添加
```
table=True
```
使用
model_validate()
进行转换 - 实现模式与表模型之间的转换

使用
sqlmodel_update()
进行部分更新 - 传递

exclude_unset=True

给

model_dump()

始终使用
Field()
定义约束 - 主键、索引、外键、默认值
使用
Annotated
依赖项 - 简洁、可复用的会话注入
使用生命周期事件创建表 - 替代已弃用的
```
@app.on_event
```
为频繁查询的列添加索引 - 使用
```
Field(index=True)
```
开发阶段启用
echo=True
- 查看生成的 SQL 查询

Common Issues

常见问题

Issue	Solution
Missing `table=True`	Add `table=True` to models that need DB tables
Circular imports	Use `TYPE_CHECKING` and string annotations for relationships
Session already closed	Ensure session is still open when accessing lazy-loaded relationships
Migration not detecting changes	Use `compare_type=True` in Alembic env.py

问题	解决方案
缺少 `table=True`	为需要数据库表的模型添加 `table=True`
循环导入	对于关系，使用 `TYPE_CHECKING` 和字符串注解
会话已关闭	访问延迟加载的关系时确保会话仍处于打开状态
迁移未检测到变更	在 Alembic 的 env.py 中设置 `compare_type=True`