Peewee ORM动态切换数据库引擎时JSONField通用实现方法问询

核心思路：基于Peewee字段扩展机制，动态适配数据库类型 #

Peewee支持通过字段的db_type()方法及绑定事件动态调整行为，我们可以利用这一点实现通用JSONField，无需针对不同数据库写冗余代码。

1. 基础通用JSONField实现 #

继承Field类，根据当前绑定的数据库类型自动切换底层JSON实现：

from peewee import Field, JSONField as PeeweeJSONField
from peewee import PostgresqlDatabase, SqliteDatabase

class UniversalJSONField(Field):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 预初始化两种数据库对应的JSONField实例
        self._sqlite_field = PeeweeJSONField(*args, **kwargs)
        self._postgres_field = PeeweeJSONField(*args, **kwargs)

    def db_type(self, context):
        # 根据数据库类型返回对应字段定义
        db = context.database
        if isinstance(db, SqliteDatabase):
            return self._sqlite_field.db_type(context)
        elif isinstance(db, PostgresqlDatabase):
            return self._postgres_field.db_type(context)
        raise ValueError("Unsupported database type")

    def python_value(self, value):
        # 统一处理数据库值转Python对象
        db = self.model._meta.database
        if isinstance(db, SqliteDatabase):
            return self._sqlite_field.python_value(value)
        elif isinstance(db, PostgresqlDatabase):
            return self._postgres_field.python_value(value)
        return value

    def db_value(self, value):
        # 统一处理Python对象转数据库存储格式
        db = self.model._meta.database
        if isinstance(db, SqliteDatabase):
            return self._sqlite_field.db_value(value)
        elif isinstance(db, PostgresqlDatabase):
            return self._postgres_field.db_value(value)
        return value

2. 结合动态绑定机制优化 #

如果你的项目通过.bind()动态切换数据库，可以利用Peewee的绑定钩子，在绑定阶段直接替换字段为对应数据库的原生实现：

from peewee import Model

def on_model_bind(model_class, database):
    # 遍历模型字段，替换通用JSONField为数据库原生实现
    for name, field in model_class._meta.fields.items():
        if isinstance(field, UniversalJSONField):
            if isinstance(database, SqliteDatabase):
                model_class._meta.fields[name] = field._sqlite_field
            elif isinstance(database, PostgresqlDatabase):
                model_class._meta.fields[name] = field._postgres_field

# 给所有模型注册绑定钩子
Model.register_bind_hook(on_model_bind)

3. 关键注意事项 #

• 用数据库vendor属性替代类型判断：可以用database.vendor == 'sqlite'或'postgres'替代实例类型检查，适配性更强。
• 封装通用JSON查询方法：SQLite与Postgres的JSON查询语法差异较大（比如Postgres支持jsonb路径查询），建议封装跨数据库的查询工具函数，避免直接写数据库特定SQL。
• 双数据库测试覆盖：针对两种数据库分别编写测试用例，确保字段读写、查询行为一致。

内容的提问来源于stack exchange，提问作者Chris Mutel