如何将 Pydantic 类型与 sqlalchemy 和 asyncpg 一起使用

Question

使用：Sqlalchemy 1.4 和 Pydantic v1

Sqlalchemy + asyncpg 查询：

async def get_api_key(
    db_session: AsyncSession,
    api_key_id: ApiKeyId,
) -> ApiKey | None:
    if api_key_id is None and auth0_id is None:
        return None
    result = await db_session.execute(
        select(ApiKey).filter(ApiKey.id == api_key_id,)
    )
    return result.scalar_one_or_none()

Sqlalchemy 模型：

class ApiKey(Base):
    id: Column = Column(
        UUID(as_uuid=True),
        primary_key=True,
    )

Pydantic 型号：

class ApiKeyId(ConstrainedStr):
   # Validate value here

class ApiKey(BaseModel):
    api_key_id: ApiKeyId

当使用

psycopg2

时，效果很好，但是当使用

asyncpg

时，会出现以下错误：

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

最简单的解决方案是将

str()

添加到查询中的

ApiKey.id

中。但是，有更好的方法吗？

Answer 1

在过滤器中使用键的

id

属性对我有用，尽管使用的是更新版本的 SQLAlchemy 和 Pydantic：

async def get_api_key(
    db_session: AsyncSession,
    api_key_id: ApiKeyId,
) -> ApiKey | None:
    if api_key_id is None and auth0_id is None:
        return None
    result = await db_session.execute(
        select(ApiKey).filter(ApiKey.id == api_key_id.id,)
    )
    return result.scalar_one_or_none()