やりたいこと

いわゆる1:Nの関係にあるデータ(売上・売上明細など)をFastAPI経由で作成する必要があり、その方法をメモします。

製品コードなどの画面入力項目を主キーにしている場合は特に問題がないのですが、
連番やuuidなどを主キーにしている場合、事前(リクエスト前)にキー項目の値を把握することができず、データ作成時にAttributeError: 'dict' object has no attribute '_sa_instance_state'というような例外が発生してしまいます。

実現方法の概要

crud処理の中で、pydanticスキーマのdictをアンパック(**schema_create.dict())せず、key=valueを直接指定する
sqlalchemy.orm.Session.flushを利用して親の主キーを取得し、子データを作成

※FastAPIの勉強を始めたばかりですので、もっと良いやり方がある場合はご教示いただけると大変嬉しいです

詳細：ソースコード

SQLAlchemyモデル
- Sales: 売上データ
- SalesMeisai: 売上明細データ

(本来売上データは商品マスター、取引先マスターといったデータを持つすべきなのですが、シンプルにするため省略しています)

models.py

from sqlalchemy import Column, Date, DateTime, Float, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm.decl_api import DeclarativeMeta
from sqlalchemy_utils import UUIDType

ModelBase: DeclarativeMeta = declarative_base()

def now():
    return datetime.now()


class Sales(ModelBase):
    """売上"""

    __tablename__ = "t_sales"
    pk = Column(UUIDType(binary=False), primary_key=True, default=uuid.uuid4)
    # 伝票番号
    slip_id = Column(String(500), nullable=False)
    # 取引日
    torihiki_date = Column(Date, nullable=False)
    post_code = Column(String(10), nullable=True)
    address1 = Column(String(2000), nullable=False, default="")
    address2 = Column(String(2000), nullable=False, default="")
    address3 = Column(String(2000), nullable=False, default="")
    phone_number = Column(String(100), nullable=True, default="")
    fax_number = Column(String(100), nullable=True, default="")
    created_at = Column(DateTime, nullable=True, default=now)
    created_by = Column(String, nullable=True, default="")
    updated_at = Column(DateTime, nullable=True, default=now)
    updated_by = Column(String, nullable=True, default="")


class SalesMeisai(ModelBase):
    """売上明細"""

    __tablename__ = "t_sales_meisai"

    pk = Column(UUIDType(binary=False), primary_key=True, default=uuid.uuid4)
    sales_header_id = Column(
        "sales_header_id", ForeignKey("t_sales.pk", onupdate="CASCADE", ondelete="CASCADE")
    )
    # 売上明細番号
    detail_no = Column(Integer, nullable=False, default=1)
    # 数量
    quantity = Column(Float(asdecimal=True), nullable=False, default=0)
    # 売上額
    sales_amount = Column(Float(asdecimal=True), nullable=False, default=0)
    created_at = Column(DateTime, nullable=True, default=now)
    created_by = Column(String, nullable=True, default="")
    updated_at = Column(DateTime, nullable=True, default=now)
    updated_by = Column(String, nullable=True, default="")

pydanticスキーマ

schemas/sales.py

from pydantic import BaseModel
from app.db.models import SalesMeisai


class SalesBase(BaseModel):
    class Config:
        orm_mode = True


class SalesMeisaiDetail(SalesBase):
    detail_no: int
    quantity: float
    sales_amount: float


class SalesCreate(SalesBase):
    slip_id: str
    invoice_number: str
    torihiki_date: datetime.date
    post_code: str
    address1: str
    address2: str
    address3: str
    phone_number: str
    fax_number: str
    notes_in_house: str
    notes_customer: str
    # ここはpydanticのBaseModelを継承したクラスを指定する
    meisai: list[SalesMeisaiDetail] = []

CRUD

crud/sales.py

from sqlalchemy import select
from sqlalchemy.orm import Session
from app.api.schemas.sales import SalesCreate
from app.db.models import Sales, SalesMeisai
def create_sales(db: Session, sales_create: SalesCreate):
    sales = Sales(
        slip_id=sales_create.dict().get("slip_id"),
        invoice_number=sales_create.dict().get("invoice_number"),
        torihiki_date=sales_create.dict().get("torihiki_date"),
        post_code=sales_create.dict().get("post_code"),
        address1=sales_create.dict().get("address1"),
        address2=sales_create.dict().get("address2"),
        address3=sales_create.dict().get("address3"),
        phone_number=sales_create.dict().get("phone_number"),
        fax_number=sales_create.dict().get("fax_number"),
    )

    db.add(sales)
    # ここでflushし、sales:ヘッダのuuidを取得
    db.flush()

    meisai_list = sales_create.dict().get("meisai")
    for meisai_data in meisai_list:
        sales_meisai = SalesMeisai(
            sales_header_id=sales.pk,
            detail_no=meisai_data["detail_no"],
            product=meisai_data["product"],
            quantity=meisai_data["quantity"],
            sales_amount=meisai_data["sales_amount"],
        )
        db.add(sales_meisai)
    db.commit()
    db.refresh(sales)
    return sales

router

router/sales.py

from app.api import dependencies as deps
from app.api.cruds import sales as sales_crud
from app.api.schemas import sales as sales_schemas
from fastapi import APIRouter, Depends, status

router = APIRouter()
@router.post(
    "/",
    response_model=sales_schemas.SalesCreate,
    status_code=status.HTTP_201_CREATED,
)
async def sales_create(sales: sales_schemas.SalesCreate, db=Depends(deps.get_db)):
    return sales_crud.create_sales(db, sales)

main.py

from fastapi import FastAPI
app = FastAPI()
app.include_router(sales_router, prefix="/api/sales", tags=["sales"])

alembicでマイグレーションを実行

$ alembic revision --autogenerate -m "add_sales_data"
$ alembic upgrade head

apiを実行し、正常終了することを確認

これで安心してフロントエンド側からコールすることができます。