如何为SQLAlchemy中的嵌套数据类创建自定义JSON映射

我想用SQLAlchemy的命令式映射持久化Python数据类Person，这个类里有个字段引用了另一个City类。City类只是两个字典的包装器，我希望把它以非规范化的方式存储为JSON列。

示例类代码：

@dataclass
class Person:
    name: str
    city: City


@dataclass
class City:
    property_a: dict
    property_b: dict

期望的数据库存储形式：

+--------+-----------------------------------------------------------------+
|  name  |                              city                               |
+--------+-----------------------------------------------------------------+
| aaron  | {property_a: {some_value: 1}, property_b: {another_value: 2}}   |
| bob    | {property_a: {some_value: 10}, property_b: {another_value: 20}} |
+--------+-----------------------------------------------------------------+

当前的表定义与映射代码：

person_table = Table(
    "persons",
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("name", String),
    Column("city", JSON)
)

mapper_registry.map_imperatively(
    Person,
    person_table
)

运行时报错：Object of type City is not JSON serializable。需要为mapper_registry添加自定义序列化/反序列化逻辑，让SQLAlchemy知道如何将City类转换为嵌套字典及反向转换，同时不确定该方案是否可行。

核心是给city字段添加自定义类型转换器，让SQLAlchemy明确City对象与JSON格式的转换规则，这是SQLAlchemy处理自定义类型映射的标准方案，完全可行。

步骤1：实现自定义类型转换器 #

继承SQLAlchemy的TypeDecorator，重写两个核心方法：

• process_bind_param：将Python对象序列化为数据库可存储的JSON格式
• process_result_value：将数据库返回的JSON反序列化为Python对象

from sqlalchemy import TypeDecorator, JSON
from dataclasses import asdict, dataclass

@dataclass
class City:
    property_a: dict
    property_b: dict

class CityJSON(TypeDecorator):
    impl = JSON

    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        # 将City对象转为字典，供JSON列存储
        return asdict(value)

    def process_result_value(self, value, dialect):
        if value is None:
            return None
        # 将数据库返回的字典转回City对象
        return City(**value)

步骤2：修改表定义，使用自定义类型 #

将原表中的JSON类型替换为自定义的CityJSON：

from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.orm import registry

mapper_registry = registry()

person_table = Table(
    "persons",
    mapper_registry.metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("name", String),
    Column("city", CityJSON)  # 替换为自定义类型
)

步骤3：执行命令式映射 #

现在执行映射即可正常工作：

@dataclass
class Person:
    name: str
    city: City

mapper_registry.map_imperatively(Person, person_table)

效果验证 #

创建实例并操作数据库时，转换逻辑会自动执行：

# 创建实例
city = City(property_a={"some_value": 1}, property_b={"another_value": 2})
person = Person(name="aaron", city=city)

# 存入数据库：city自动序列化为字典形式的JSON
# 查询数据：数据库返回的JSON自动转回City对象

方案合理性说明 #

该方案是SQLAlchemy的标准实践，优势如下：

• 类型转换逻辑封装在转换器中，与业务代码解耦
• 兼容命令式与声明式两种映射方式
• 完全满足非规范化存储需求，无需额外创建关联表

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