Как использовать типы Pydantic с sqlalchemy и asyncpg ⇐ Python

[Anonymous]

Использование: 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,
)

Пидантические модели:

Код: Выделить всё

class ApiKeyId(ConstrainedStr):
# Validate value here

class ApiKey(BaseModel):
api_key_id: ApiKeyId

При использовании psycopg2 это работает отлично, но при использовании asyncpg возникает следующая ошибка: error = DataError('неверный ввод для аргумента запроса $1: ApiKeyId (abcdefgh... (Ожидается юникод, получен ApiKeyId)')
Самое простое решение — просто добавить str() в ApiKey.id в запросе. Однако есть ли лучший способ сделать это?

Подробнее здесь: https://stackoverflow.com/questions/790 ... nd-asyncpg