Python SQL Alchemy Nested Object Insert

Question

I am using a Python and SQLAlchemy to perform CRUD operations on my oracle database.

I have the below models:

class ServiceOrderModel(BaseModel):
    service_order_id: int | None = None
    service_order_number: str | None = None
    service_type: str
    service_order_items: Optional[List[ServiceOrderItemModel]] = None

    class Config:
        from_attributes = True

class ServiceOrderItemModel(BaseModel):
    service_order_item_id: Optional[int] = None
    service_order_id: Optional[int] = None
    component_id: Optional[int] = None
    component: Optional[ComponentModel] = None
    total_qty: Optional[float]
    unit_price: Optional[float]
    service_order_items_received: Optional[List[ServiceOrderItemReceiveModel]] = None

    class Config:
        from_attributes = True


class ComponentModel(BaseModel):
    component_id: Optional[int] = None
    component_serial_number: str = None
    component_name: Optional[str] = None

    class Config:
        from_attributes = True

When a user creates a service order they also have to create multiple items (One to many). Each item also has a (One to one) relationship with component.

When I have the ServiceOrderModel, what is the best way to perform an insert into the service_order, service_order_item, and component table with this single object.

Would it be best to split it up? How would I do that as I would need the service_order_id generated from the service_order table to perform an insert into the item table. I have my table def's below as well.

class ServiceOrder(BaseModel):
    __tablename__ = "service_order"

    service_order_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    service_order_number: Mapped[Optional[str]] = mapped_column(
        String(42),
        server_default=text(
            """
    CASE "SERVICE_TYPE" 
        WHEN 'NEW' THEN 'SO'||TO_CHAR("SERVICE_ORDER_ID") 
        WHEN 'REPAIR' THEN 'RO'||TO_CHAR("SERVICE_ORDER_ID") 
        ELSE NULL 
    END
"""
        ),
    )
    service_type: Mapped[str] = mapped_column(String(50))
    service_order_item: Mapped[List["ServiceOrderItem"]] = relationship(
        "ServiceOrderItem", back_populates="service_order"
    )


class ServiceOrderItem(BaseModel):
    __tablename__ = "service_order_item"

    service_order_item_id: Mapped[int] = mapped_column(
        primary_key=True, autoincrement=True
    )
    component_id: Mapped[Optional[int]] = mapped_column(
        ForeignKey("component.component_id")
    )
    service_order_id: Mapped[int] = mapped_column(
        ForeignKey("service_order.service_order_id")
    )
    total_qty: Mapped[Optional[int]]
    unit_price: Mapped[Optional[float]]
    component: Mapped["Component"] = relationship(
        "Component", back_populates="service_order_item"
    )
    service_order: Mapped["ServiceOrder"] = relationship(
        "ServiceOrder", back_populates="service_order_item"
    )


class Component(BaseModel):
    __tablename__ = "component"

    component_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    component_name: Mapped[str] = mapped_column(String(500))
    component_serial_number: Mapped[str] = mapped_column(String(250), unique=True)
    service_order_item: Mapped[List["ServiceOrderItem"]] = relationship(
        "ServiceOrderItem", back_populates="component"
    )

Answer 1

I was able to insert all 3 objects with one insert/add statement. Below is the function I created to do this.

async def create_purchase_order(
    session: AsyncSession, new_service_order: CreatePurchaseOrderRequest
) -> ServiceOrderModel:
    async with session:
        service_order_dict = new_service_order.model_dump(by_alias=True)
        service_order_dict["service_order_item"] = [
            DBServiceOrderItem(
                component=DBComponent(**item.pop("component", {})),
                **item,
            )
            if item.get("component")
            else DBServiceOrderItem(
                **item,
            )
            for item in service_order_dict.pop("service_order_items")
        ]

        service_order_orm = DBServiceOrder(**service_order_dict)

        session.add(service_order_orm)
        await session.commit()

        return service_order_orm.service_order_id