Postgres自定义triple_lang类型在Pydantic+SQLModel栈中返回列表问题求助

1. 修正字段不匹配的核心问题 #

你定义的Postgres复合类型triple_lang字段为(en text, fr text, es text)，但Python侧TripleLang类的字段是en, ru, uk，字段名完全不对应，这是导致类型转换异常的关键之一，先统一字段：

from typing import Optional
from sqlmodel import SQLModel

class TripleLang(SQLModel, table=False):
    en: Optional[str] = None
    fr: Optional[str] = None
    es: Optional[str] = None

    class Config:
        from_attributes = True

2. 配置SQLAlchemy复合类型与自动转换 #

方式一：在模型层实现自动转换 #

利用SQLModel的from_orm方法，在查询后自动将PG返回的元组/列表转为TripleLang对象：

from sqlalchemy_utils import CompositeType
from sqlmodel import SQLModel, Field
from typing import Optional

# 定义与PG复合类型匹配的CompositeType
triple_lang_type = CompositeType(
    'triple_lang',
    [
        ('en', SQLModel.sqlalchemy_column_type(str)),
        ('fr', SQLModel.sqlalchemy_column_type(str)),
        ('es', SQLModel.sqlalchemy_column_type(str)),
    ]
)

class DataExtended(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: Optional[TripleLang] = Field(sa_column=triple_lang_type())
    description: Optional[TripleLang] = Field(sa_column=triple_lang_type())

    @classmethod
    def from_orm(cls, obj):
        instance = super().from_orm(obj)
        # 把元组/列表转换为TripleLang对象
        if isinstance(instance.title, (tuple, list)):
            instance.title = TripleLang(en=instance.title[0], fr=instance.title[1], es=instance.title[2])
        if isinstance(instance.description, (tuple, list)):
            instance.description = TripleLang(en=instance.description[0], fr=instance.description[1], es=instance.description[2])
        return instance

方式二：自定义SQLAlchemy类型转换器 #

创建自定义类型，继承CompositeType并实现结果处理器，自动完成类型转换：

from sqlalchemy_utils import CompositeType
from sqlalchemy import TypeDecorator
from sqlmodel import SQLModel, Field
from typing import Optional

class TripleLangType(TypeDecorator):
    impl = CompositeType(
        'triple_lang',
        [
            ('en', SQLModel.sqlalchemy_column_type(str)),
            ('fr', SQLModel.sqlalchemy_column_type(str)),
            ('es', SQLModel.sqlalchemy_column_type(str)),
        ]
    )

    def result_processor(self, dialect, coltype):
        base_processor = self.impl.result_processor(dialect, coltype)
        def process(value):
            if value is None:
                return None
            # 将PG返回的元组转为TripleLang对象
            return TripleLang(en=value[0], fr=value[1], es=value[2])
        return process

# 在表模型中使用自定义类型
class DataExtended(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: Optional[TripleLang] = Field(sa_column=TripleLangType())
    description: Optional[TripleLang] = Field(sa_column=TripleLangType())

3. 确保psycopg2复合类型注册正确 #

使用psycopg2注册复合类型时，要指定完整的类型名（带schema，比如public.triple_lang），并全局生效：

import psycopg2
from psycopg2.extensions import register_composite

# 建立数据库连接后执行注册
conn = psycopg2.connect("你的数据库连接字符串")
register_composite('public.triple_lang', conn.cursor(), globally=True)

4. 验证查询结果 #

执行查询后检查字段类型：

from sqlmodel import Session, select

with Session(engine) as session:
    data = session.exec(select(DataExtended)).first()
    print(type(data.title))  # 应输出 <class '__main__.TripleLang'>
    print(data.title.en)  # 可直接访问字段

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