やりたいこと
いわゆる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を実行し、正常終了することを確認
これで安心してフロントエンド側からコールすることができます。