Преобразовать Oracle sys_guid в строку и обратно в приложении на FastAPI при помощи Pydantic

Question

Необходима помощь с Pydantic. Есть приложение на FastAPI.

Model.py

class ReportGroup(Base):
    __tablename__ = 'site_report_groups'
    id = sa.Column(RAW(16), primary_key=True, server_default=sa.text("SYS_GUID()"))
    title = sa.Column('title', sa.String(200))
    categories = relationship('ReportCategory', secondary='site_report_category_group', back_populates='groups')
    reports = relationship('Report', secondary='site_report_group', back_populates='groups')


class ReportCategory(Base):
    __tablename__ = 'site_report_categories'
    id = sa.Column(RAW(16), primary_key=True, server_default=sa.text("SYS_GUID()"))
    title = sa.Column('title', sa.String(200))
    groups = relationship('ReportGroup', secondary='site_report_category_group', back_populates='categories')


class Report(Base):
    __tablename__= 'site_reports'
    id = sa.Column(RAW(16), primary_key=True, server_default=sa.text("SYS_GUID()"))
    title = sa.Column('title', sa.String(200))
    query = sa.Column('query', sa.CLOB)
    is_deleted = sa.Column('is_deleted', sa.Boolean, default=False)
    chartstype = sa.Column('charts_type', sa.Integer)
    legend = sa.Column('legend', sa.String(255))
    datatype = sa.Column('data_type', sa.Integer)
    columns = sa.Column('columns', sa.String(255))
    groups = relationship('ReportGroup', secondary='site_report_group', back_populates='reports')


site_report_category_group = sa.Table('site_report_category_group', Base.metadata,
    sa.Column('id', RAW(16), primary_key=True, server_default=sa.text("SYS_GUID()")),
    sa.Column('group_id', sa.ForeignKey('site_report_groups.id')),# primary_key=True),
    sa.Column('categories_id', sa.ForeignKey('site_report_categories.id'))#, primary_key=True)
)


site_report_group = sa.Table('site_report_group', Base.metadata,
    sa.Column('id', RAW(16), primary_key=True, server_default=sa.text("SYS_GUID()")),
    sa.Column('group_id', sa.ForeignKey('site_report_groups.id')),# primary_key=True),
    sa.Column('reports_id', sa.ForeignKey('site_reports.id'))#, primary_key=True)
)

Schema.py

class Report(BaseModel):
    id:str
    title:str
    is_deleted:Optional[bool]
    chartstype:int
    datatype:int
    columns:Optional[str]

    class Config:
        orm_mode=True


class Group(BaseModel):
    id:str 
    title:str
    reports:List[Report]

    class Config:
        orm_mode=True


class Category(BaseModel):
    id:str
    title:str
    groups:List[Group]
    
    class Config:
        orm_mode=True

Делаем запрос к БД Oracle через FastAPI

def get_reports(db):
    result = []
    for row in db.query(model.Report).all():
        row.id = row.id.hex()
        result.append(row)
    return result

И это срабатывает. Код выполняется без ошибок. Но когда я выполняю этот запрос

def get_categories(db):
    result=[]
    for categ in db.query(model.ReportCategory).all():
        # categ.id = categ.id.hex()
        result.append(categ)
        for group in categ.groups:
            # group.id = group.id.hex()
            result.append(group)
            for report in group.reports:
                # report.id = report.id.hex()
                result.append(report)
    return result

Он тоже срабатывает но получается такой вывод:

Traceback (most recent call last):
  File "D:\Work\Python\KolyanVenv\lib\site-packages\uvicorn\protocols\http\h11_impl.py", line 373, in run_asgi
    result = await app(self.scope, self.receive, self.send)
  File "D:\Work\Python\KolyanVenv\lib\site-packages\uvicorn\middleware\proxy_headers.py", line 75, in __call__
    return await self.app(scope, receive, send)
  File "D:\Work\Python\KolyanVenv\lib\site-packages\fastapi\applications.py", line 208, in __call__
    await super().__call__(scope, receive, send)
  File "D:\Work\Python\KolyanVenv\lib\site-packages\starlette\applications.py", line 112, in __call__
    await self.middleware_stack(scope, receive, send)
  File "D:\Work\Python\KolyanVenv\lib\site-packages\starlette\middleware\errors.py", line 181, in __call__
    raise exc
  File "D:\Work\Python\KolyanVenv\lib\site-packages\starlette\middleware\errors.py", line 159, in __call__
    await self.app(scope, receive, _send)
  File "D:\Work\Python\KolyanVenv\lib\site-packages\starlette\exceptions.py", line 82, in __call__
    raise exc
  File "D:\Work\Python\KolyanVenv\lib\site-packages\starlette\exceptions.py", line 71, in __call__
    await self.app(scope, receive, sender)
  File "D:\Work\Python\KolyanVenv\lib\site-packages\starlette\routing.py", line 656, in __call__
    await route.handle(scope, receive, send)
  File "D:\Work\Python\KolyanVenv\lib\site-packages\starlette\routing.py", line 259, in handle
    await self.app(scope, receive, send)
  File "D:\Work\Python\KolyanVenv\lib\site-packages\starlette\routing.py", line 61, in app
    response = await func(request)
  File "D:\Work\Python\KolyanVenv\lib\site-packages\fastapi\routing.py", line 234, in app
    response_data = await serialize_response(
  File "D:\Work\Python\KolyanVenv\lib\site-packages\fastapi\routing.py", line 137, in serialize_response
    raise ValidationError(errors, field.type_)
pydantic.error_wrappers.ValidationError: 4 validation errors for Report
response -> 0 -> id
  'utf-8' codec can't decode byte 0xe6 in position 1: invalid continuation byte (type=value_error.unicodedecode)
response -> 1 -> id
  'utf-8' codec can't decode byte 0xa8 in position 0: invalid start byte (type=value_error.unicodedecode)
response -> 2 -> id
  'utf-8' codec can't decode byte 0xf3 in position 0: invalid continuation byte (type=value_error.unicodedecode)
response -> 3 -> id
  'utf-8' codec can't decode byte 0xd2 in position 1: invalid continuation byte (type=value_error.unicodedecode)

Таким образом понятно, что дело в том, что Pydantic не может декодировать получаемый из Oracle sys_guid в строку. Я пробовал использовать @validators, json_encoders, но так и не добился результата. Пробовал непосредственно в схеме менять данныеid:str=str.hex(), так же безрезультатно. У кого больше опыта с Pydantic подскажет как мне изменить файл schema.py чтобы у меня все заработало?

UPD:Пробовал использовать from pydantic import UUID4. Вот этот запрос работает

@router.get('/reports', response_model=List[schema.Report])
def get_reports(db):
    return db.query(model.Report).all()

а этот нет

@router.get('/report_categories', response_model=List[schema.Category])
def get_categories(db):
    return db.query(model.ReportCategory).all()

Хотя я написал версию для Sqlite и 2 запроса указанных выше , которые замечательно работают именно на Sqlite, но надо на Oracle и обязательно с sys_guid на котором последний запрос не работает

Answer 1

После долгих размышлений пришел к такому решению. Файл schema.py удаляем. Файл models.py оставляем без изменений. Далее файл с роутингами (routers.py):

from db import db_reports
...
@router.get('/reports')#, response_model=List[Report])
def report_all(db: Session = Depends(get_db)):
    return list(map(db_reports.ReportDto, db.query(mod.Report).all()))


@router.get('/report-categories') #response_model=List[Category])
def category_all(db: Session = Depends(get_db)):
    return list(map(db_reports.ReportCategoryDto, db.query(mod.ReportCategory).all()))

Далее файл взаимодействия с БД Oracle:

from db import models as mod
...
class ReportDto : 
    id:str
    title:str
    datatype:int
    chartsType:int
    columns:str

    def __init__(self, model:mod.Report) -> None:
        self.id = model.id
        self.title = model.title
        self.datatype = model.datatype
        self.chartstype = model.chartstype
        self.columns = model.columns

class ReportGroupDto : 
    id:str
    title:str
    reports:List[ReportDto]

    def __init__(self, model:mod.ReportGroup) -> None:
        self.id = model.id
        self.title = model.title
        self.Reports = list(map(ReportDto, model.reports))

class ReportCategoryDto : 
    id:str
    title:str
    groups:List[ReportGroupDto]

    def __init__(self, model:mod.ReportCategory) -> None:
        self.id = model.id
        self.title = model.title
        self.groups = list(map(ReportGroupDto, model.groups))

Спасибо всем, кто помогал прийти к этому решению.

БЛОГ НА HUSL

Преобразовать Oracle sys_guid в строку и обратно в приложении на FastAPI при помощи Pydantic

Ответы (1 шт):