我正在尝试使用 sqlalchemy 连接两个表,我想要实现的查询是:
SELECT *
FROM service_order
JOIN vendor ON service_order.vendor_id = vendor.vendor_id
WHERE service_order.service_order_id = 898;
下面我当前的 sqlalchemy 语句仅生成一个对象
from typing import List, Sequence, Dict, Union
from sqlalchemy import select
from sqlalchemy import func
from sqlalchemy.orm import Session, joinedload, lazyload, selectinload
from api.model import CreateServiceOrderRequest
from db.model import ServiceOrder as DBServiceOrder
from db.model import ServiceOrderItem as DBServiceOrderItem
from db.model import Vendor as DBVendor
from db.engine import DatabaseSession as AsyncSession
from ..exceptions import DBRecordNotFoundException, InvalidPropertyException
from core.domains.service_order.service_order_model import ServiceOrderModel
async def get_service_order_by_id(
session: AsyncSession, service_order_id: int
) -> ServiceOrderModel:
async with session:
statement = (
select(DBServiceOrder)
.options(
joinedload(DBServiceOrder.service_order_item).joinedload(
DBServiceOrderItem.service_order_item_receive
),
)
.join(DBVendor, DBServiceOrder.vendor_id == DBVendor.vendor_id)
.where(DBServiceOrder.service_order_id == service_order_id)
)
result = await session.scalars(statement)
service_order = result.first()
return service_order
以下是我的 service_order 和供应商的数据库模型。
from datetime import datetime
from typing import Optional, List
from sqlalchemy import func, ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .base_model import BaseModel
class ServiceOrder(BaseModel):
__tablename__ = "service_order"
service_order_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
vendor_id: Mapped[float] = mapped_column(ForeignKey("vendor.vendor_id"))
vendor: Mapped["Vendor"] = relationship(
"Vendor", lazy="selectin", back_populates="service_order"
)
service_order_item: Mapped[List["ServiceOrderItem"]] = relationship(
"ServiceOrderItem", back_populates="service_order"
)
from datetime import datetime
from typing import Optional, List
from sqlalchemy import func, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .base_model import BaseModel
class Vendor(BaseModel):
__tablename__ = "vendor"
vendor_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
company_name: Mapped[Optional[str]] = mapped_column(String(200))
component: Mapped[List["Component"]] = relationship(back_populates="vendor")
motor: Mapped[List["Motor"]] = relationship(foreign_keys="Motor.vendor_id")
motor_field_track: Mapped[List['MotorFieldTrack']] = relationship('MotorFieldTrack', back_populates='vendor')
service_order: Mapped[List['ServiceOrder']] = relationship('ServiceOrder', back_populates='vendor')
motor_order: Mapped[List['MotorOrder']] = relationship('MotorOrder', back_populates='vendor')
我尝试过 joinloading、lazyloading、joininload 的各种组合,但似乎没有任何效果。它不是创建包含所有列的对象,而是继续创建供应商对象而不是该表中的列。如何返回一个包含两个表中所有列的对象?如果我计划从第二个表中选择哪些列就更好了,我该如何实现这一目标?
通常,您只想在获取数千条记录或其他内容时执行此操作,因为您几乎是在针对 ORM 工作。请注意,您必须使用
execute
而不是 scalars
来访问包含其他“列”的元组。另外,如果您不使用 ORM 功能,那么您也不会想使用 joinedload
。
async def get_service_order_by_id(
session: AsyncSession, service_order_id: int
) -> ServiceOrderModel:
async with session:
statement = (
select(DBServiceOrder, DBVendor.company_name)
.join(DBVendor, DBServiceOrder.vendor_id == DBVendor.vendor_id)
.where(DBServiceOrder.service_order_id == service_order_id)
)
result = await session.execute(statement)
service_order, vender_company_name = result.first()
传统的ORM方式是这样的。使用
joinload
可同时填充 .vendor
对象并将其附加到 server_order
对象上,所有操作均通过一个 SQL
查询进行。从常规的SQL
来看,这听起来很疯狂,但最终在大多数情况下你最终都会手动进行这种积累和分组:
async def get_service_order_by_id(
session: AsyncSession, service_order_id: int
) -> ServiceOrderModel:
async with session:
statement = (
select(DBServiceOrder)
.join(DBServiceOrder.vendor)
.where(DBServiceOrder.service_order_id == service_order_id)
.options(joinedload(DBServiceOrder.vendor))
)
result = await session.scalars(statement)
service_order = result.first()
# without joinedload above this would trigger a SQL
# select in the background to load this
vendor = server_order.vendor