はじめに
この記事は JSL (日本システム技研) Advent Calendar 2024 - Qiita 20日目の記事です。
昨年は、本業多忙で書けませんでしたが、今年は無事にbirthday駆動開発（BDD）の記事を書けました
。
前回の続きでFAST APIを使った簡易CRM実装の続きを記事にします。
今回は、CRUDのビジネスロジックを中心に記事にします。
FAST APIの公式ドキュメントを参考に進めてきます。
おさらい
環境：
・macOS 12.6 Apple M1 Pro
・Python 3.12.7
現状のフォルダ構成は以下となります。
├── backend
│ ├── app
│ │ ├── __init__.py
│ │ ├── __pycache__
│ │ │ ├── __init__.cpython-312.pyc
│ │ │ ├── main.cpython-312.pyc
│ │ │ └── models.cpython-312.pyc
│ │ ├── api
│ │ │ ├── __init__.py
│ │ │ ├── __pycache__
│ │ │ │ ├── __init__.cpython-312.pyc
│ │ │ │ └── main.cpython-312.pyc
│ │ │ ├── main.py
│ │ │ └── routes
│ │ │ ├── __init__.py
│ │ │ ├── __pycache__
│ │ │ │ ├── __init__.cpython-312.pyc
│ │ │ │ ├── customer.cpython-312.pyc
│ │ │ │ └── hello.cpython-312.pyc
│ │ │ ├── customer.py
│ │ │ └── hello.py
│ │ ├── core
│ │ │ ├── __init__.py
│ │ │ ├── config.py
│ │ │ └── db.py
│ │ ├── main.py
│ │ ├── models.py
│ │ └── tests
│ └── requirements.txt
├── poetry.lock
└── pyproject.toml
CRUDのビジネスロジック実装
DIで、SQLModelのSessionを渡す
前回の記事で、create_engineにて作成したDBのengineをFAST APIの
DIを使用して各CRUDの関数(path operation)に渡していきます。
app/core/db.py
from typing import Annotated
from fastapi import Depends
from sqlmodel import Session
def get_session():
with Session(engine) as session:
yield session
SessionDep = Annotated[Session, Depends(get_session)]
app/api/routes/customer.py
from app.core.db import SessionDep
# session: SessionDepを引数に追加
@router.get("/customers", response_model=List[Customer])
async def read_customers(session: SessionDep):
return customers
@router.post("/customers", response_model=Customer)
async def create_customer(customer: Customer, session: SessionDep) :
return customer
@router.put("/customers/{customer_id}", response_model=Customer)
async def update_customer(customer_id: int, customer: Customer, session: SessionDep):
return customer
@router.delete("/customers/{customer_id}")
async def delete_customer(customer_id: int, session: SessionDep):
return {"message": "Customer deleted"}
SQLModelを使用してビジネスロジックを実装
チュートリアルを参考にし、一部公式ドキュメントで補足しながら実装していきます。
app/api/routes/customer.py
@router.get("/customers", response_model=List[Customer])
async def read_customers(
session: SessionDep,
offset: int = 0,
limit: Annotated[int, Query(le=100)] = 100,
):
customers = session.exec(select(Customer).offset(offset).limit(limit)).all()
return customers
@router.post("/customers", response_model=Customer)
async def create_customer(customer: Customer, session: SessionDep) :
session.add(customer)
session.commit()
session.refresh(customer)
return customer
@router.put("/customers/{customer_id}", response_model=Customer)
async def update_customer(customer_id: int, customer: Customer, session: SessionDep):
print(f"update_customer start")
db_customer: Customer = session.get(Customer, customer_id)
if not db_customer:
raise HTTPException(status_code=404, detail="Customer not found")
customer_data = customer.model_dump(exclude_unset=True)
db_customer.sqlmodel_update(customer_data)
session.add(db_customer)
session.commit()
session.refresh(db_customer)
return db_customer
@router.delete("/customers/{customer_id}")
async def delete_customer(customer_id: int, session: SessionDep):
customer = session.get(Customer, customer_id)
if not customer:
raise HTTPException(status_code=404, detail="Customer not found")
session.delete(customer)
session.commit()
return {"ok": True}
まとめ
3回に渡って、FAST APIのチュートリアルと合わせながらCRUD実装を記事にしてみました。
普段、DjangoとDRFでの開発が多いだけにFAST APIによる実装は新鮮でした。
まだまだ、簡易CRMを作るに当たってFAST APIを使い倒したいので、最後に今後やっていきたいことを自分用のメモとしてまとめました。
今後やっていくこと
- SQLModelを使った複数ModelのQueryを書く
- SQLModelを使ってValidationを書く
- テストを書く
- 認証・認可の実装
- セキュリティ対応