Блог

Я использую фреймворк FastAPI с Asyncio. Я недавно перешел на Postgresql из MySQL и поменялся aiomysql asyncpg соответствующим образом.

В MySQL я мог бы извлечь это из базы данных, и оно автоматически было бы приведено в схему Pydantic.

 async def get_device(device_id: str) -> Device:
    query = DeviceTable.select().where(DeviceTable.c.id == device_id)
    return await db.fetch_one(query)
 

Схема Pydantic:

 class Device(BaseModel):
    id: str
    type: DeviceType
    created_at: datetime
    last_login_at: datetime
    is_banned: bool
    ...
 

Однако при использовании Postgres это преобразование, похоже, больше не происходит.

 device: Device = await crud_device.get_device(body.device_id)
 

Ибо device.is_banned я получаю сообщение об ошибке {AttributeError}'Record' object has no attribute 'is_banned' .

Объект устройства в этом случае является объектом записи <databases.backends.postgres.Record object at 0x10d2c9860> вместо схемы Pydantic.

Я все еще могу получить доступ к таким значениям:

device['is_banned']

Почему поведение FastAPI изменилось только потому, что я выбрал другую базу данных? Я ничего не могу найти в документации по этому поводу.

Обновить

Модель:

 DeviceTable = sqlalchemy.Table(
    "device",
    metadata,
    Column("id", String(36), primary_key=True),
    Column("type", Enum(DeviceType), nullable=False),
    Column(
        "created_at",
        TIMESTAMP(),
        nullable=False,
        default=func.now(),
        server_default=func.now(),
    ),
    Column("last_login_at", TIMESTAMP(), nullable=True),
    Column("expires_at", TIMESTAMP()),
    Column("account_type", Enum(AccountType), nullable=False),
    Column("strikes", Integer(), nullable=True),
    Column(
        "is_banned",
        Boolean(),
        nullable=False,
        server_default=sql.false(),
        default=sql.false(),
    ),
    Column("account_id", ForeignKey("account.id"), nullable=True, index=True),
)
 

Кажется, я нашел обходной путь. Я должен явно поместить запись в схему Pydantic.

 async def get_device(device_id: str) -> Device:
    query = DeviceTable.select().where(DeviceTable.c.id == device_id)
    record = await db.fetch_one(query)
    return Device(**record)
 

Это довольно странно, что это не происходит автоматически. Мне нужно изменить код, потому что изменилась базовая база данных. Хуже всего то, что, когда запись не будет найдена, актерский состав рухнет. Следовательно, мне приходится вносить более повторяющиеся изменения:

 async def get_device(device_id: str) -> Optional[Device]:
    query = DeviceTable.select().where(DeviceTable.c.id == device_id)
    record = await db.fetch_one(query)
    if record:
        return Device(**record)
    else:
        return None