Как использовать тип стрелки в схеме ответа FastAPI?

Question

Как использовать тип стрелки в схеме ответа FastAPI?

Я хочу использовать Arrow введите FastAPI ответ, потому что я использую его уже в SQLAlchemy модель (спасибо sqlalchemy_utils).

Я подготовил небольшой автономный пример с минимальным приложением FastAPI. Я ожидаю, что это приложение вернется product1 данные из базы данных.

К сожалению, приведенный ниже код дает исключение:

      Exception has occurred: FastAPIError
Invalid args for response field! Hint: check that <class 'arrow.arrow.Arrow'> is a valid pydantic field type

      import sqlalchemy
import uvicorn
from arrow import Arrow
from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy import Column, Integer, Text, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy_utils import ArrowType

app = FastAPI()

engine = sqlalchemy.create_engine('sqlite:///db.db')
Base = declarative_base()

class Product(Base):
    __tablename__ = "product"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(Text, nullable=True)
    created_at = Column(ArrowType(timezone=True), nullable=False, server_default=func.now())

Base.metadata.create_all(engine)


Session = sessionmaker(bind=engine)
session = Session()

product1 = Product(name="ice cream")
product2 = Product(name="donut")
product3 = Product(name="apple pie")

session.add_all([product1, product2, product3])
session.commit()


class ProductResponse(BaseModel):
    id: int
    name: str
    created_at: Arrow

    class Config:
        orm_mode = True
        arbitrary_types_allowed = True


@app.get('/', response_model=ProductResponse)
async def return_product():

    product = session.query(Product).filter(Product.id == 1).first()

    return product

if __name__ == "__main__":
    uvicorn.run(app, host="localhost", port=8000)

requirements.txt:

      sqlalchemy==1.4.23
sqlalchemy_utils==0.37.8
arrow==1.1.1
fastapi==0.68.1
uvicorn==0.15.0

Эта ошибка уже обсуждалась в тех проблемах FastAPI:

Один из возможных обходных путей - добавить этот код (исходный код ):

      from pydantic import BaseConfig
BaseConfig.arbitrary_types_allowed = True

Достаточно поставить чуть выше @app.get('/'..., но его можно поставить и раньше app = FastAPI()

Проблема с этим решением заключается в том, что конечная точка GET выводит:

      // 20210826001330
// http://localhost:8000/

{
  "id": 1,
  "name": "ice cream",
  "created_at": {
    "_datetime": "2021-08-25T21:38:01+00:00"
  }
}

вместо желаемого:

      // 20210826001330
// http://localhost:8000/

{
  "id": 1,
  "name": "ice cream",
  "created_at": "2021-08-25T21:38:01+00:00"
}

7

sqlalchemy fastapi pydantic sqlalchemy-utils arrow-python

Источник

user8896457 26 авг '21 в 00:57

4 ответа

Другие вопросы по тегам sqlalchemy fastapi pydantic sqlalchemy-utils arrow-python

user4260095 02 дек '21 в 09:04 2021-12-02 09:04 · Answer 1 · 2021-12-02 09:04

Добавьте пользовательскую функцию с помощью@validatorдекоратор, который возвращает желаемый _datetimeобъекта:

      class ProductResponse(BaseModel):
    id: int
    name: str
    created_at: Arrow

    class Config:
        orm_mode = True
        arbitrary_types_allowed = True

    @validator("created_at")
    def format_datetime(cls, value):
        return value._datetime

Проверено на локальном, вроде работает:

      $ curl -s localhost:8000 | jq
{
  "id": 1,
  "name": "ice cream",
  "created_at": "2021-12-02T08:25:10+00:00"
}

user8896457 14 май '22 в 00:46 2022-05-14 00:46 · Answer 2 · 2022-05-14 00:46

Решение состоит в том, чтобы установить pydantic ENCODERS_BY_TYPEпоэтому он знает, как преобразовать объект Arrow, чтобы его можно было принять в формате json:

      from arrow import Arrow
from pydantic.json import ENCODERS_BY_TYPE
ENCODERS_BY_TYPE |= {Arrow: str}

Параметр BaseConfig.arbitrary_types_allowed = Trueтакже необходимо.

Результат:

      // 20220514022717
// http://localhost:8000/

{
  "id": 1,
  "name": "ice cream",
  "created_at": "2022-05-14T00:20:11+00:00"
}

Полный код:

      import sqlalchemy
import uvicorn
from arrow import Arrow
from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy import Column, Integer, Text, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy_utils import ArrowType

from pydantic.json import ENCODERS_BY_TYPE
ENCODERS_BY_TYPE |= {Arrow: str}

from pydantic import BaseConfig
BaseConfig.arbitrary_types_allowed = True

app = FastAPI()

engine = sqlalchemy.create_engine('sqlite:///db.db')
Base = declarative_base()

class Product(Base):
    __tablename__ = "product"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(Text, nullable=True)
    created_at = Column(ArrowType(timezone=True), nullable=False, server_default=func.now())

Base.metadata.create_all(engine)


Session = sessionmaker(bind=engine)
session = Session()

product1 = Product(name="ice cream")
product2 = Product(name="donut")
product3 = Product(name="apple pie")

session.add_all([product1, product2, product3])
session.commit()


class ProductResponse(BaseModel):
    id: int
    name: str
    created_at: Arrow

    class Config:
        orm_mode = True
        arbitrary_types_allowed = True


@app.get('/', response_model=ProductResponse)
async def return_product():

    product = session.query(Product).filter(Product.id == 1).first()

    return product

if __name__ == "__main__":
    uvicorn.run(app, host="localhost", port=8000)

user16124869 06 апр '23 в 12:37 2023-04-06 12:37 · Answer 3 · 2023-04-06 12:37

Недавно я столкнулся с подобной проблемой, и ответ, предоставленный @Karol Zlot, кажется устаревшим - FastAPI выдавал ошибку схемы JSON:

      ValueError: Value not declarable with JSON Schema, field: name='created_at' type=ArrowType required=True

Кажется, что код ниже работает:

      import datetime

class ArrowType(datetime):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v):
        return v._datetime

class Domain(DomainBase):
    id: int
    created_at: ArrowType
    updated_at: ArrowType

user14305096 21 июн '22 в 17:38 2022-06-21 17:38 · Answer 4 · 2022-06-21 17:38

Вот пример кода, где вам не нужноclass Configи может работать для любого типа, создав свой собственный подкласс с валидаторами:

      from psycopg2.extras import DateTimeTZRange as DateTimeTZRangeBase
from sqlalchemy.dialects.postgresql import TSTZRANGE
from sqlmodel import (
    Column,
    Field,
    Identity,
    SQLModel,
)

from pydantic.json import ENCODERS_BY_TYPE

ENCODERS_BY_TYPE |= {DateTimeTZRangeBase: str}


class DateTimeTZRange(DateTimeTZRangeBase):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v):
        if isinstance(v, str):
            lower = v.split(", ")[0][1:].strip().strip()
            upper = v.split(", ")[1][:-1].strip().strip()
            bounds = v[:1] + v[-1:]
            return DateTimeTZRange(lower, upper, bounds)
        elif isinstance(v, DateTimeTZRangeBase):
            return v
        raise TypeError("Type must be string or DateTimeTZRange")

    @classmethod
    def __modify_schema__(cls, field_schema):
        field_schema.update(type="string", example="[2022,01,01, 2022,02,02)")


class EventBase(SQLModel):
    __tablename__ = "event"
    timestamp_range: DateTimeTZRange = Field(
        sa_column=Column(
            TSTZRANGE(),
            nullable=False,
        ),
    )


class Event(EventBase, table=True):
    id: int | None = Field(
        default=None,
        sa_column_args=(Identity(always=True),),
        primary_key=True,
        nullable=False,
    )

ссылка на выпуск Github: https://github.com/tiangolo/sqlmodel/issues/235#issuecomment-1162063590