SQLAlchemy 将父表与子表连接起来，其中子表需要来自另一个表的信息

问题描述 投票：0回答：1

我正在构建一个 FastAPI 应用程序，并努力使用 SQLAlchemy v2 连接多个表以实现特定的查询输出格式。我有以下三个模型：

class MaterialReceipt(Base, AuditMixins):
    material_receipt_id = Column(Integer, primary_key=True, index=True)
    pic = Column(String)
    is_visible = Column(Boolean)
    material_receipt_lines = relationship("MaterialReceiptLine", backref='material_receipt')


class MaterialReceiptLine(Base, AuditMixins):
    material_receipt_line_id = Column(Integer, primary_key=True, index=True)
    material_id = Column(Integer, ForeignKey("material.material_id"))
    material_receipt_id = Column(Integer, ForeignKey("material_receipt.material_receipt_id"))
    quantity = Column(Float)
    exp_date = Column(Date)

class Material(Base, AuditMixins):
    material_id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    sku = Column(String)
    location = Column(String)
    low_stock_threshold = Column(Float)


Material Receipt
是父级，
MaterialReceiptLine
是子级。 
MaterialReceiptLine
通过 
Material
链接到 
material_id
。我想查询所有的物料收据，结果应该是以下格式：

[
    {
        "material_receipt_id": ...,
        "pic": ...,
        "is_visible": ...
        "material_receipt_lines": [
            {
                "material_receipt_line_id": ...,
                "name": ...,
                "sku": ...,
                "quantity": ...,
                "exp_date": ...
            }, ...
        ]
    }, ...
]

我知道为了加入 

Material Receipt
和 
Material Receipt Line
我可以使用 sqlalchemy 尝试以下代码：

        material_receipts = (
            (self.db
             .query(MaterialReceiptModel)
             .join(MaterialReceiptLineModel)
             .options(contains_eager(MaterialReceiptModel.lines))
             )
            .filter(MaterialReceiptModel.is_visible)
            .order_by(MaterialReceiptModel.created_at.desc())
            .all()
        )

但是，我不知道如何加入

Material
和
Material Receipt Line
来实现所需的输出格式。有办法做到吗？

谢谢！

python join sqlalchemy
1个回答
0
投票

回答您的问题并不容易，因为您没有提供回答问题所需的所有信息，或者您提供的信息是矛盾的。

例如：

  • 您使用什么数据库？ （因为有些选项并非在所有数据库系统中都可用）
  • 您的数据库中的表名称是什么？ （我们必须猜测）
  • 什么是
    AuditMixins
    ，它与您的问题相关吗？
  • 您提供的查询似乎引用了
    MaterialReceiptModel
    、
    MaterialReceiptLineModel
    、
    MaterialReceiptModel
    。它们没有在您的示例中的任何地方定义，但您已经定义了 
    MaterialReceipt
    、
    MaterialReceiptLine
    、
    Material
    。我们是否应该猜测它们应该是相同的东西？

总而言之，您没有向我们提供最小的可重现示例

我们只能猜测您想做什么。您要求的输出是分层的，但查询无法直接为您提供答案。

所以这是我从你的问题中得到的理解：

import random
import string
import datetime
import json

from sqlalchemy import Column, Integer, String, Boolean, ForeignKey, Float, Date, create_engine
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# Defining the classes mapped
Base = declarative_base()

class MaterialReceipt(Base):
    __tablename__ = "material_receipt"

    id = Column(Integer, primary_key=True)
    pic = Column(String)
    is_visible = Column(Boolean)
    material_receipt_lines = relationship("MaterialReceiptLine", backref="material_receipt")

class MaterialReceiptLine(Base):
    __tablename__ = "material_receipt_lines"

    id = Column(Integer, primary_key=True, index=True)
    material_id = Column(Integer, ForeignKey("material.id"))
    material_receipt_id = Column(Integer, ForeignKey("material_receipt.id"))
    quantity = Column(Float)
    exp_date = Column(Date)
    material = relationship("Material")

class Material(Base):
    __tablename__ = "material"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    sku = Column(String)
    location = Column(String)
    low_stock_threshold = Column(Float)

# Tools used to generate example random data
def generate_random_string():
    return "".join(random.choices(string.ascii_letters, k=20))

location_examples = (
    "Paris",
    "London",
    "New York",
    "Sydney",
    "Tokyo",
    "Los Angeles",
    "Moscow",
    "Berlin",
    "San Francisco" 
)

amount_of = {
    "receipts": lambda: range(random.randint(2, 4)),
    "materials": lambda: range(random.randint(2, 6))
}

# Creating random objects and saving them to our example database
engine = create_engine("sqlite:///:memory:")

Base.metadata.create_all(engine)

Session = sessionmaker()
Session.configure(bind=engine)

session = Session()

receipts = (
    MaterialReceipt(
        pic=generate_random_string(),
        is_visible=random.choice((True, False)),
        material_receipt_lines=[
            MaterialReceiptLine(
                quantity=random.uniform(0, 5000),
                exp_date=(datetime.date.today() + datetime.timedelta(days=random.randint(1, 90))),
                material=Material(
                    name=generate_random_string(),
                    sku=generate_random_string(),
                    location=random.choice(location_examples),
                    low_stock_threshold=random.uniform(10, 20)
                )
            )
            for also_not_used in amount_of["materials"]()
        ]
    )
    for not_used in amount_of["receipts"]()
)

session.add_all(receipts)
session.commit()
session.close()

# The answer is here:

# Retreving the objects from the database
another_session = Session()

output = [
    {
        "material_receipt_id": receipt.id,
        "pic": receipt.pic,
        "is_visible": receipt.is_visible,
        "material_receipt_lines": [
            {
                "material_receipt_line_id": line.id,
                "name": line.material.name,
                "sku": line.material.sku,
                "quantity": line.quantity,
                "exp_date": line.exp_date.isoformat()
            }
            for line in receipt.material_receipt_lines
        ]
    }
    for receipt 
    in another_session.query(MaterialReceipt).filter(MaterialReceipt.is_visible).all()
]

another_session.close()

# Displaying the output in the format asked
print(json.dumps(output, indent=4))

我在这个示例中使用了内存中的 SQLite，但如果您使用其他东西，情况可能会有所不同。我利用了这样一个事实：通过定义对象之间的关系，我可以轻松获得 

MaterialReceipt
、
MaterialReceiptLine
、
Material
对象之间的链接。

我必须提到，您要求的输出看起来很像 JSON 格式，因此我利用了 

json
python 模块作为输出格式。我还利用了这样一个事实：自 Python 3.7 以来，字典保留插入顺序，以便 
output
字典在转换为 JSON 对象时具有正确的顺序值。

但是您提到正在使用 SQLAlchemy v2。但你的代码风格被认为是遗留的。从现在开始，您应该使用更现代的 SQLAlchemy 习惯用法。您至少应该看看 SQLAlchemy ORM 快速入门 以了解应该如何完成工作。

所以这个答案看起来像：

import datetime
import random
import string
import json

from typing import List

from sqlalchemy import Boolean, ForeignKey, Date, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

# Defining the classes mapped
class Base(DeclarativeBase):
    pass

class MaterialReceipt(Base):
    __tablename__ = "material_receipt"

    id: Mapped[int] = mapped_column(primary_key=True)
    pic: Mapped[str]
    is_visible: Mapped[bool] = mapped_column(Boolean())
    material_receipt_lines: Mapped[List["MaterialReceiptLine"]] = relationship(back_populates="material_receipt", cascade="all, delete-orphan")

    def __repr__(self) -> str:
        receipt_lines = "\n\t".join(f"{x!r}" for x in self.material_receipt_lines)
        return f"Material(id={self.id!r}, pic={self.pic!r}, is_visible={self.is_visible!r}, material_receipt_lines=[\n\t{receipt_lines}\n])"

class MaterialReceiptLine(Base):
    __tablename__ = "material_receipt_lines"

    id: Mapped[int] = mapped_column(primary_key=True)
    material_id: Mapped[int] = mapped_column(ForeignKey("material.id"))
    material_receipt_id: Mapped[int] = mapped_column(ForeignKey("material_receipt.id"))
    quantity: Mapped[float]
    exp_date: Mapped[datetime.date] = mapped_column(Date)
    
    material_receipt: Mapped["MaterialReceipt"] = relationship(back_populates="material_receipt_lines")
    material: Mapped["Material"] = relationship()

    def __repr__(self) -> str:
        return f"MaterialReceiptLine(id={self.id!r}, quantity={self.quantity!r}, exp_date={self.exp_date!r}, material=\n\t\t{self.material!r})"

class Material(Base):
    __tablename__ = "material"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    sku: Mapped[str]
    location: Mapped[str]
    low_stock_threshold: Mapped[float]

    def __repr__(self) -> str:
        return f"Material(id={self.id!r}, name={self.name!r}, sku={self.sku!r}, location={self.location!r}, low_stock_threshold={self.low_stock_threshold})"

# Tools used to generate example random data
def generate_random_string():
    return "".join(random.choices(string.ascii_letters, k=20))

location_examples = (
    "Paris",
    "London",
    "New York",
    "Sydney",
    "Tokyo",
    "Los Angeles",
    "Moscow",
    "Berlin",
    "San Francisco" 
)

amount_of = {
    "receipts": lambda: range(random.randint(2, 4)),
    "materials": lambda: range(random.randint(2, 6))
}

# Creating random objects and saving them to our example database
engine = create_engine("sqlite://")

Base.metadata.create_all(engine)

with Session(engine) as session:
    receipts = (
        MaterialReceipt(
            pic=generate_random_string(),
            is_visible=random.choice((True, False)),
            material_receipt_lines=[
                MaterialReceiptLine(
                    quantity=random.uniform(0, 5000),
                    exp_date=(datetime.date.today() + datetime.timedelta(days=random.randint(1, 90))),
                    material=Material(
                        name=generate_random_string(),
                        sku=generate_random_string(),
                        location=random.choice(location_examples),
                        low_stock_threshold=random.uniform(10, 20)
                    )
                )
                for also_not_used in amount_of["materials"]()
            ]
        )
        for not_used in amount_of["receipts"]()
    )

    session.add_all(receipts)
    session.commit()

# The answer is here:

# Retreving the objects from the database
with Session(engine) as another_session:
    stmt = select(MaterialReceipt).where(MaterialReceipt.is_visible)

    output = [
        {
            "material_receipt_id": receipt.id,
            "pic": receipt.pic,
            "is_visible": receipt.is_visible,
            "material_receipt_lines": [
                {
                    "material_receipt_line_id": line.id,
                    "name": line.material.name,
                    "sku": line.material.sku,
                    "quantity": line.quantity,
                    "exp_date": line.exp_date.isoformat()
                }
                for line in receipt.material_receipt_lines
            ]
        }
        for receipt in another_session.scalars(stmt)
    ]

# Displaying the output in the format asked
print(json.dumps(output, indent=4))
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.