如何在SQLAlchemy 1.4与asyncpg中正确使用Pydantic类型？

问题背景 #

使用SQLAlchemy 1.4 + Pydantic v1时，切换到asyncpg驱动后执行查询出现错误：

DataError('invalid input for query argument $1: ApiKeyId(abcdefgh... (Expected unicode, got ApiKeyId)')

psycopg2下代码正常运行，临时方案是手动用str()转换ApiKeyId参数，以下是更优的解决方案：

推荐方案：让ApiKeyId继承str #

Pydantic v1的ConstrainedStr本身继承自str，但如果你的ApiKeyId仅继承ConstrainedStr，可能因asyncpg严格的类型检查导致识别失败。修改ApiKeyId类同时继承str和ConstrainedStr，确保实例能被当作原生字符串处理：

from pydantic import ConstrainedStr
from uuid import UUID

class ApiKeyId(str, ConstrainedStr):
    # 保留你的验证规则，例如UUID格式校验、长度限制
    min_length = 36
    max_length = 36

    @classmethod
    def validate(cls, value):
        validated_value = super().validate(value)
        # 示例：校验是否为有效UUID
        try:
            UUID(validated_value)
        except ValueError:
            raise ValueError("ApiKeyId必须是有效的UUID字符串")
        return validated_value

修改后，ApiKeyId实例直接具备字符串特性，无需手动转换即可传入SQLAlchemy查询，asyncpg能正确识别参数类型。

备选方案：注册SQLAlchemy自定义类型转换器 #

若不想修改Pydantic模型，可通过SQLAlchemy的类型装饰器自动转换ApiKeyId为UUID类型：

from sqlalchemy import TypeDecorator, UUID
from sqlalchemy.ext.declarative import Base

class ApiKeyIdType(TypeDecorator):
    impl = UUID(as_uuid=True)

    def process_bind_param(self, value, dialect):
        if isinstance(value, ApiKeyId):
            return UUID(str(value))
        return value

# 更新SQLAlchemy模型的id列定义
class ApiKey(Base):
    id: Column = Column(
        ApiKeyIdType,
        primary_key=True,
    )

该方案让SQLAlchemy在绑定查询参数时自动处理ApiKeyId类型，适配asyncpg的参数要求。

临时过渡方案：统一转换工具函数 #

如果暂时无法修改模型或类型，可封装工具函数避免重复编写str()：

def to_db_identifier(api_key_id: ApiKeyId | None) -> str | None:
    return str(api_key_id) if api_key_id else None

# 查询时调用工具函数
result = await db_session.execute(
    select(ApiKey).filter(ApiKey.id == to_db_identifier(api_key_id),)
)

方案对比 #

• 推荐方案侵入性最低，代码简洁，同时保留Pydantic的验证能力，是最优选择。
• 备选方案适合需要复杂类型转换控制的场景，例如多自定义类型统一处理。
• 临时方案仅作为过渡，长期维护性不如前两者。

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