如何使用Pydantic模式与SQLalchemy关系分别列

Question

我有4个表：硬件，softwarename，softwareVersion和软件。

软件表具有与softwarename表和SoftwareVersion表的一对多关系。最后，硬件模型具有与软件表的一对多关系。

我正在尝试使用Pydantic Sc​​hema获得模型关系中的特定列。

现在我将获得此输出：

[
  {
    "id": 1,
    "hostname": "hostname2",
    "softwares": [
      {
        "id": 1,
        "software_name": {
          "id": 1,
          "name": "nginx"
        },
        "software_version": {
          "id": 1,
          "version": "2.9"
        }
      },
      {
        "id": 2,
        "software_name": {
          "id": 2,
          "name": "vim"
        },
        "software_version": {
          "id": 2,
          "version": "0.3"
        }
      },
      {
        "id": 3,
        "software_name": {
          "id": 3,
          "name": "apache"
        },
        "software_version": {
          "id": 3,
          "version": "1.0"
        }
      }
    ]
  }
]

但是我期望的是输出：

[
  {
    "id": 1,
    "hostname": "hostname2",
    "softwares": [
      {
        "id": 1,
        "name": "nginx",
        "version": "2.9"
      },
      {
        "id": 2,
        "name": "vim",
        "version": "0.3"
      },
      {
        "id": 3,
        "name": "apache",
        "version": "1.0"
      }
    ]
  }
]

我有文件main.py：

   
import uvicorn
from typing import Any, Iterator, List, Optional
from faker import Faker
from fastapi import Depends, FastAPI
from pydantic import BaseModel
from sqlalchemy import Column, ForeignKey, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, sessionmaker, relationship
from faker.providers import DynamicProvider

software_name = DynamicProvider(
     provider_name="software_name",
     elements=["bash", "vim", "vscode", "nginx", "apache"],
)


software_version = DynamicProvider(
     provider_name="software_version",
     elements=["1.0", "2.9", "1.1", "0.3", "2.0"],
)


hardware = DynamicProvider(
     provider_name="hardware",
     elements=["hostname1", "hostname2", "hostname3", "hostname4", "hostname5"],
)

fake = Faker()

# then add new provider to faker instance
fake.add_provider(software_name)
fake.add_provider(software_version)
fake.add_provider(hardware)


engine = create_engine("sqlite:///.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=True, autoflush=True, bind=engine)

Base = declarative_base(bind=engine)


class Software(Base):
    __tablename__ = 'software'

    id = Column(Integer, primary_key=True)
    hardware_id = Column(Integer, ForeignKey('hardware.id'))
    name_id = Column(Integer, ForeignKey('software_name.id'))
    version_id = Column(Integer, ForeignKey('software_version.id'))

    software_name = relationship('SoftwareName', backref='software_name')
    software_version = relationship('SoftwareVersion',
                                    backref='software_version')


class SoftwareName(Base):
    __tablename__ = 'software_name'

    id = Column(Integer, primary_key=True)
    name = Column(String)


class SoftwareVersion(Base):
    __tablename__ = 'software_version'

    id = Column(Integer, primary_key=True)
    version = Column(String)


class Hardware(Base):
    __tablename__ = "hardware"

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String, nullable=False)

    softwares = relationship(Software)


Base.metadata.drop_all()
Base.metadata.create_all()

class BaseSchema(BaseModel):
    id: int

    class Config:
        orm_mode = True


class SoftwareNameSchema(BaseSchema):
    name: str


class SoftwareVersionSchema(BaseSchema):
    version: str


class SoftwareSchema(BaseSchema):
    software_name: SoftwareNameSchema
    software_version:  SoftwareVersionSchema


class HardwareOut(BaseSchema):
    hostname: str
    softwares: List[SoftwareSchema]


app = FastAPI()


@app.on_event("startup")
def on_startup() -> None:
    session = SessionLocal()

    for _ in range(10):
        software_list = []
        for _ in range(3):
            sn = SoftwareName(name=fake.software_name())
            sv = SoftwareVersion(version=fake.software_version())
            s = Software(software_name=sn, software_version=sv)
            software_list.append(s)

        h = Hardware(hostname=fake.hardware(), softwares=software_list)
        session.add(h)
        session.flush()

    session.close()


def get_db() -> Iterator[Session]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.get("/hardwares", response_model=List[HardwareOut])
def get_hardwares(db: Session = Depends(get_db)) -> Any:
    return [HardwareOut.from_orm(hardware) for hardware in db.query(Hardware).all()]

如何更改hardwareout架构以返回我期望的东西？

Answer 1

我终于得到了我想要的答案。

我添加了2个更改以获取它：

1. 使用Union从键入属性 lib software> software_name e software> software_version代码>这样：

2. 添加pydantic 验证器为每个字段更改返回的值，例如：

from typing import Union
from pydantic import validator

...

class SoftwareSchema(BaseSchema):
    software_name: Union[str, SoftwareNameSchema]
    software_version:  Union[str, SoftwareVersionSchema]

    @validator('software_name')
    def name_to_str(cls, v, values, **kwargs):
        return v.name if not isinstance(v, str) else v

    @validator('software_version')
    def version_to_str(cls, v, values, **kwargs):
        return v.version if not isinstance(v, str) else v

...

答案是：

[
  {
    "id": 1,
    "hostname": "hostname2",
    "softwares": [
      {
        "id": 1,
        "software_name": "nginx",
        "software_version": "2.9"
      },
      {
        "id": 2,
        "software_name": "vim",
        "software_version": "0.3"
      },
      {
        "id": 3,
        "software_name": "apache",
        "software_version": "1.0"
      }
    ]
  }
]

更新：

作为改进，我为每个属性添加一个别名，以获得更好的语义响应。因此，我将software_name更改为name和software_version 版本。像这样：

from typing import Union
from pydantic import validator

...

class SoftwareSchema(BaseSchema):
    software_name: Union[str, SoftwareNameSchema] = Field(None, alias="name")
    software_version:  Union[str, SoftwareVersionSchema] = Field(None, alias="version")

    @validator('software_name')
    def name_to_str(cls, v, values, **kwargs):
        return v.name if not isinstance(v, str) else v

    @validator('software_version')
    def version_to_str(cls, v, values, **kwargs):
        return v.version if not isinstance(v, str) else v

...