一、プロジェクトの作成
PyCharmでFastAPIプロジェクトを作成します：
二、技術実装
- データベース操作
実際のデータベースCRUD操作をカプセル化し、SQLAlchemyを使用して実装します。コードはbase.pyに記述します：
from typing import Type, TypeVar, List, Generic, Optional, Any, Union, Iterable
from sqlalchemy.orm import Session
# ジェネリック型変数を定義
EntityType = TypeVar("EntityType")
class DataAccessLayer(Generic[EntityType]):
def __init__(self, entity_type: Type[EntityType]):
"""
:param entity_type: SQLAlchemyのエンティティクラス（例：User）
"""
self.entity_type = entity_type
# --- 検索 ---
def retrieve(self, db_session: Session, identifier: Any) -> Optional[EntityType]:
"""IDに基づいて単一のレコードを取得"""
return db_session.get(self.entity_type, identifier)
def retrieve_multiple(
self,
db_session: Session,
*,
offset: int = 0,
limit: int = 100,
conditions: list = None, # [User.age > 18, User.name.like("%A%")]のようなフィルタを受け取る
sort_by: Optional[Union[Any, Iterable[Any]]] = None, # 単一または複数のソートフィールドをサポート
) -> List[EntityType]:
"""複数のレコードを取得（ページネーションをサポート）"""
query = db_session.query(self.entity_type)
if conditions:
query = query.filter(*conditions)
if sort_by is not None:
# 単一のフィールド（例：User.id）が渡された場合、リストに変換して処理
if not isinstance(sort_by, (list, tuple)):
sort_by = [sort_by]
# *演算子を使用してリストを展開
query = query.order_by(*sort_by)
return query.offset(offset).limit(limit).all()
# --- 作成 ---
def create(
self,
db_session: Session,
*,
input_data: Union[dict, EntityType] # 辞書またはインスタンス化されたモデルを受け入れる
) -> EntityType:
"""レコードを作成"""
# 1. データを辞書形式に統一
if isinstance(input_data, dict):
create_payload = input_data
else:
# モデルインスタンスが渡された場合、有効なフィールドを抽出
create_payload = {
c.name: getattr(input_data, c.name)
for c in input_data.__table__.columns
if getattr(input_data, c.name) is not None # 未設定のフィールドを除外
}
# 2. 核心ロジック：アンパッキングを使用してモデルをインスタンス化
# これにより手動でsetattrをループする必要がなく、パフォーマンスが向上
db_object = self.entity_type(**create_payload)
db_session.add(db_object)
try:
db_session.commit()
db_session.refresh(db_object) # データベースで自動生成されたIDやデフォルト値を取得
except Exception as e:
db_session.rollback() # 競合（例：一意索引エラー）が発生した場合、ロールバックが必要
raise e
return db_object
# --- 更新 ---
def update(
self,
db_session: Session,
*,
db_object: EntityType = None,
input_data: Union[dict, EntityType] # 辞書またはモデルインスタンスを受け入れる
) -> EntityType:
"""レコードを更新"""
# 1. 辞書形式に統一
if isinstance(input_data, dict):
update_payload = input_data
else:
# SQLAlchemyモデルの場合、辞書に変換（内部状態フィールドを除外）
update_payload = {
c.name: getattr(input_data, c.name)
for c in input_data.__table__.columns
}
# 2. 更新ロジックを実行
for field in update_payload:
if hasattr(db_object, field) and update_payload[field] is not None:
setattr(db_object, field, update_payload[field])
db_session.add(db_object)
db_session.commit()
db_session.refresh(db_object)
return db_object
# --- 削除 ---
def delete(self, db_session: Session, *, identifier: int) -> EntityType:
"""レコードを削除"""
obj = db_session.query(self.entity_type).get(identifier)
db_session.delete(obj)
db_session.commit()
return obj
- データベース接続
pymysqlを使用してMySQLデータベースに接続し、データベース操作を実装します。コードはdatabase.pyに記述します：
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 1. データベース接続情報を定義（SQLiteを例として、MySQL/Postgresの場合はURLを置き換える）
DB_USER = "root"
DB_PASSWORD = "your_password"
DB_HOST = "127.0.0.1"
DB_PORT = "3306"
DB_NAME = "test_db"
SQLALCHEMY_DATABASE_URL = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4"
# 2. エンジンを作成
# pool_recycle: 接続を自動的にリサイクルし、MySQLのデフォルト8時間の切断による"MySQL server has gone away"を防止
# pool_size: コネクションプールのサイズ
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
pool_size=10,
max_overflow=20,
pool_recycle=3600,
pool_pre_ping=True
)
# 3. セッションファクトリを作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 4. 基本マッピングクラスを作成
Base = declarative_base()
# データベース接続を取得するユーティリティ関数（依存注入でよく使用）
def get_db_session():
db = SessionLocal()
try:
yield db
finally:
db.close()
- シンプルなモデル
従業員モデルemployee.pyを使用します：
from datetime import datetime
from typing import Annotated, Optional
from pydantic import BaseModel, ConfigDict
from pydantic import PlainSerializer
from sqlalchemy import Column, String, DateTime, DOUBLE
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class StaffMember(Base):
__tablename__ = "staff_member"
id = Column(String, primary_key=True)
name = Column(String)
salary = Column(DOUBLE)
creation_time = Column(DateTime)
# 1. グローバルで一般的な日付タイプを定義
CustomDatetime = Annotated[
datetime,
PlainSerializer(lambda v: v.strftime("%Y-%m-%d %H:%M:%S"), return_type=str)
]
# --- レスポンスモデル（Pydantic v2）---
class StaffMemberSchema(BaseModel):
id: Optional[str] = None
name: Optional[str] = None
salary: Optional[float] = None
creation_time: Optional[CustomDatetime] = None
model_config = ConfigDict(from_attributes=True)
- CRUDの実装
従業員テーブルのCRUDを実装します。コードはmain.pyに記述します：
from datetime import datetime
from typing import Any, Optional
from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from db.base import DataAccessLayer
from db.employee import StaffMember, StaffMemberSchema
from db.database import get_db_session
import json
class CustomJSONResponse(JSONResponse):
def render(self, content: any) -> bytes:
return json.dumps(
content,
ensure_ascii=False,
allow_nan=False,
indent=None,
separators=(",", ":"),
# ここですべての日付オブジェクトを処理
default=lambda obj: obj.strftime("%Y-%m-%d %H:%M:%S") if isinstance(obj, datetime) else str(obj),
).encode("utf-8")
app = FastAPI(default_response_class=CustomJSONResponse)
# インスタンス化
staff_member_service = DataAccessLayer(StaffMember)
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/staff_member/retrieve", response_model=StaffMemberSchema)
async def get_staff_member(identifier: Any, db_session: Session = Depends(get_db_session)):
return staff_member_service.retrieve(db_session, identifier=identifier)
@app.post("/staff_member/list", response_model=list[StaffMemberSchema])
async def list_staff_members(filter_data: Optional[dict] = None, db_session: Session = Depends(get_db_session)):
if filter_data is not None and "name" in filter_data:
return staff_member_service.retrieve_multiple(db_session, conditions=[StaffMember.name.like("%" + filter_data["name"] + "%")])
return staff_member_service.retrieve_multiple(db_session)
@app.post("/staff_member/create", response_model=StaffMemberSchema)
async def create_staff_member(data: dict, db_session: Session = Depends(get_db_session)):
data["creation_time"] = datetime.now()
staff_member = StaffMember(**data)
return staff_member_service.create(db_session, input_data=staff_member)
@app.post("/staff_member/update", response_model=StaffMemberSchema)
async def update_staff_member(data: dict, db_session: Session = Depends(get_db_session)):
staff_member = StaffMember(**data)
db_object = staff_member_service.retrieve(db_session, identifier=staff_member.id)
if db_object is None:
raise HTTPException(status_code=404, detail="Staff member not found")
return staff_member_service.update(db_session, db_object=db_object, input_data=staff_member)
@app.post("/staff_member/remove", response_model=StaffMemberSchema)
async def remove_staff_member(data: dict, db_session: Session = Depends(get_db_session)):
return staff_member_service.delete(db_session, identifier=data["id"])
三、テスト
1、起動
8080ポートを指定して実行します：
--host 0.0.0.0 --port 8080
2、新規作成
/staff_member/createインターフェースを呼び出してデータを新規作成します：
結果：新規作成成功
3、更新
/staff_member/updateインターフェースを呼び出して従業員データを更新します
結果：更新成功
4、一覧表示
/staff_member/listインターフェースを呼び出してデータを表示します
結果：従業員リスト
5、削除
/staff_member/removeで従業員データを削除します：
結果：削除成功