Mixin 类
TIP
本页是参考文档。要看怎么把这些 Mixin 组合到自己的模型上,去 操作指南。
所有 Mixin 通过 MRO 组合到 table 模型上。MRO 顺序通常重要——见每个 Mixin 的"MRO"小节。
CachedTableBaseMixin
from sqlmodel_ext import CachedTableBaseMixin继承自 TableBaseMixin。为模型的 get() 查询添加 Redis 缓存层。
MRO:CachedTableBaseMixin 必须放在 UUIDTableBaseMixin / TableBaseMixin 之前:
class Character(CachedTableBaseMixin, CharacterBase, UUIDTableBaseMixin, table=True, cache_ttl=1800):
pass类变量:
| 名称 | 类型 | 默认值 | 说明 |
|---|---|---|---|
__cache_ttl__ | int | 3600 | 缓存 TTL(秒)。可用 cache_ttl=N 关键字设置 |
_redis_client | Any | None | Redis 客户端,必须用 configure_redis() 设置 |
on_cache_hit | Callable[[str], None] | None | None | 缓存命中回调,参数为模型名 |
on_cache_miss | Callable[[str], None] | None | None | 缓存未命中回调,参数为模型名 |
类方法:
@classmethod
def configure_redis(cls, client: Any) -> None启动时调用一次。client 是 redis.asyncio.Redis 实例(decode_responses=False)。
@classmethod
def check_cache_config(cls) -> None启动时调用一次(configure_redis() 之后)。检查所有子类配置正确性,注册 SQLAlchemy session 事件钩子。
@classmethod
async def invalidate_by_id(cls, *_ids: Any) -> None手动失效一个或多个 ID 的缓存。
@classmethod
async def invalidate_all(cls) -> None失效该模型的所有缓存(ID + 查询)。
get() 新参数:在 CRUD 方法 的 get() 之外,多出 no_cache: bool = False。
自动跳过缓存的场景:with_for_update=True、populate_existing=True、options 非空、join 非空、事务内有待失效数据、no_cache=True。
OptimisticLockMixin
from sqlmodel_ext import OptimisticLockMixin, OptimisticLockErrorMRO:OptimisticLockMixin 必须放在 UUIDTableBaseMixin / TableBaseMixin 之前。
字段:
| 字段 | 类型 | 默认值 | 数据库行为 |
|---|---|---|---|
version | int | 0 | 每次 UPDATE 自动 +1(SQLAlchemy version_id_col 机制) |
类标记:
_has_optimistic_lock: ClassVar[bool] = True让 save() / update() 知道需要处理乐观锁逻辑。
触发条件:UPDATE 时 WHERE version = ? 不匹配(影响 0 行)→ StaleDataError → 由 save() / update() 转换为 OptimisticLockError。
OptimisticLockError
class OptimisticLockError(Exception):
model_class: str | None
record_id: str | None
expected_version: int | None
original_error: Exception | Nonesave() / update() 的 optimistic_retry_count 重试耗尽后抛出。
PolymorphicBaseMixin
from sqlmodel_ext import PolymorphicBaseMixin字段:
| 字段 | 类型 | 说明 |
|---|---|---|
_polymorphic_name | Mapped[str] | 鉴别列(String、有索引)。子类自动写入;不参与 API 序列化 |
__init_subclass__ 接受的关键字参数:
| 参数 | 默认值 | 含义 |
|---|---|---|
polymorphic_on | '_polymorphic_name' | 鉴别列字段名 |
polymorphic_abstract | 自动检测 | 是否为抽象基类(含 ABC + 抽象方法时自动 True) |
类方法:
@classmethod
def _is_joined_table_inheritance(cls) -> bool
@classmethod
def get_concrete_subclasses(cls) -> list[type]
@classmethod
def get_identity_to_class_map(cls) -> dict[str, type]get_identity_to_class_map() 返回如 {'emailnotification': EmailNotification, ...}。
AutoPolymorphicIdentityMixin
from sqlmodel_ext import AutoPolymorphicIdentityMixin__init_subclass__ 接受的关键字参数:
| 参数 | 默认值 | 含义 |
|---|---|---|
polymorphic_identity | 自动生成 | 显式指定时直接用,否则用 {parent_identity}.{class_name.lower()} |
自动生成的 identity 格式是点分层级,如 'function' → 'function.codeinterpreter'。
create_subclass_id_mixin()
from sqlmodel_ext import create_subclass_id_mixin签名:
def create_subclass_id_mixin(parent_table_name: str) -> type动态生成一个 Mixin,提供指向 {parent_table_name}.id 的外键 + 主键。仅 JTI 子类需要。
MRO 要求:返回的 Mixin 必须放在继承列表第一位,让其 id 字段覆盖 UUIDTableBaseMixin 的 id。
register_sti_columns_for_all_subclasses()
from sqlmodel_ext import (
register_sti_columns_for_all_subclasses,
register_sti_column_properties_for_all_subclasses,
)签名:
def register_sti_columns_for_all_subclasses() -> None
def register_sti_column_properties_for_all_subclasses() -> NoneSTI 子类字段需要分两阶段注册到父表:
register_sti_columns_for_all_subclasses()— 在configure_mappers()之前调用register_sti_column_properties_for_all_subclasses()— 在configure_mappers()之后调用
RelationPreloadMixin
from sqlmodel_ext import RelationPreloadMixin继承此 Mixin 的类可以使用 @requires_relations 和 @requires_for_update 装饰方法。
__init_subclass__ 行为:扫描所有方法,对带有 _required_relations 元数据的方法做导入时验证(关系名拼写检查)。
实例方法:
def _is_relation_loaded(self, rel_name: str) -> bool
async def _ensure_relations_loaded(
self,
session: AsyncSession,
relations: tuple[str | QueryableAttribute, ...],
) -> None
@classmethod
def get_relations_for_method(cls, method_name: str) -> tuple
@classmethod
def get_relations_for_methods(cls, *method_names: str) -> tuple
async def preload_for(self, session: AsyncSession, *method_names: str) -> None通常你不需要直接调用这些方法——@requires_relations 装饰器会自动调用。
默认 lazy='raise_on_sql'
0.2.0 起,所有 SQLModel Relationship 字段的默认 lazy 设置为 'raise_on_sql':访问未预加载的关系立刻抛异常,而不是触发隐式同步查询。这是 MissingGreenlet 问题的最后一道安全网。