FastAPI、SQLAlchemy、PostgreSQLでRESTful APIを構築する

PythonにはFlask、Django、FastAPIなど、さまざまなWebフレームワークがあります。その中でも、FastAPIはシンプルさ、パフォーマンス、使いやすさで際立っています。

FastAPIはPydanticとStarletteの上に構築されており、組み込みの型検証と非同期サポートを提供します。FlaskやDjangoを使ったことがある方は、FastAPIが非常にシンプルであることに気づくでしょう :)

FastAPIを探索し、SQLAlchemyとPostgreSQLを使ってRESTful APIを構築してみましょう。

`FastAPI`について ⚡️

Starlette（高性能な非同期Webアプリケーション用の軽量ASGIフレームワーク）とPydantic（データモデルを定義・検証するためのデータ検証ライブラリ）上に構築されています。
Swagger UIとReDocによる即座のAPIドキュメントサポート。
高性能アプリケーション構築のための非同期サポート、依存性注入サポートなど、さまざまな機能を提供します。

FastAPI固有の機能~

FastAPI(): 自動的にOpenAPIスキーマを生成するアプリケーションインスタンスを作成
Depends(): データベースセッション、認証などの依存性注入
HTTPException: ステータスコード付きのHTTPエラーレスポンス
Pydanticモデル: 自動的なリクエスト検証とレスポンスのシリアライゼーション
response_model: レスポンス構造を定義・検証

APIを構築する 🚀

PostgreSQLデータベースを使用してストアの製品を管理するRESTful APIを構築します。このAPIは以下をサポートします：

GET /products → すべての製品を取得
POST /products → 新しい製品を作成
GET /products/:id → 特定の製品を取得
PUT /products/:id → 製品の詳細を更新
DELETE /products/:id → 製品を削除

それでは始めましょう！

0. 前提条件とセットアップ

始める前に、Python 3.8以上とDockerがインストールされていることを確認してください。プロジェクト用の新しいディレクトリを作成し、仮想環境をセットアップします：

bash

~ $ mkdir products-api && cd products-api

~/products-api $ python -m venv .venv

~/products-api $ source .venv/bin/activate  

# Windowsの場合: .venv\Scripts\activate

必要な依存関係を含むrequirements.txtファイルを作成します：

txt

annotated-doc==0.0.4
annotated-types==0.7.0
anyio==4.12.1
click==8.3.1
fastapi==0.128.0
h11==0.16.0
idna==3.11
psycopg2-binary==2.9.11
pydantic==2.12.5
pydantic-settings==2.12.0
pydantic_core==2.41.5
python-dotenv==1.2.1
SQLAlchemy==2.0.45
starlette==0.50.0
typing-inspection==0.4.2
typing_extensions==4.15.0
uvicorn==0.40.0

依存関係をインストールします：

bash

~/products-api $ pip install -r requirements.txt

1. シンプルに始める

セットアップを確認するために、最小限のFastAPIサーバーを作成しましょう。main.pyを作成します：

python

from fastapi import FastAPI

app = FastAPI(
    title="Products API",
)

@app.get("/")
def root():
    return {
        "message": "Welcome to Products API", "status": "active"
        }

サーバーを実行します：

bash

~/products-api $ uvicorn main:app --reload

http://localhost:8000 にアクセスしてAPIレスポンスを確認し、http://localhost:8000/docs にアクセスして自動生成されたインタラクティブなドキュメントを確認してください。

2. データベース設定

💡 簡単にセットアップできるよう、DockerでPostgreSQLを使用します。SQLAlchemy ORMがデータベース操作を処理します。

PostgreSQL用のdocker-compose.ymlを作成します：

yaml

services:
  db:
    image: postgres:15.4-alpine
    environment:
      - POSTGRES_USER=admin
      - POSTGRES_PASSWORD=admin123
      - POSTGRES_DB=products_db
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U admin"]
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  pgdata: {}

データベース設定用の.envファイルを作成します：

bash

DATABASE_URL=postgresql://admin:admin123@localhost:5432/products_db

データベースを起動します：

bash

~/products-api $ docker-compose up -d

3. SQLAlchemyのセットアップ

a. Pydanticを使用して環境変数を管理するsettings.pyを作成します：

python

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    database_url: str

    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()

b. データベース接続用のdatabase.pyを作成します：

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from settings import settings

engine = create_engine(settings.database_url)
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

4. データベースモデルを定義する

SQLAlchemyモデルを含むdatabase_models.pyを作成します：

python

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, Float

class Base(DeclarativeBase):
    pass

class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    description: Mapped[str] = mapped_column(String(255))
    price: Mapped[float] = mapped_column(Float)
    quantity: Mapped[int] = mapped_column(Integer)

5. Pydanticモデルを定義する

リクエスト/レスポンス検証用のmodels.pyを作成します：

python

from pydantic import BaseModel, Field
from typing import Optional

class ProductCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str = Field(..., min_length=1, max_length=255)
    price: float = Field(..., gt=0, description="Must be positive")
    quantity: int = Field(..., ge=0, description="Must be non-negative")

class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    description: Optional[str] = Field(None, min_length=1, max_length=255)
    price: Optional[float] = Field(None, gt=0)
    quantity: Optional[int] = Field(None, ge=0)

class ProductResponse(BaseModel):
    id: int
    name: str
    description: str
    price: float
    quantity: int

    class Config:
        from_attributes = True  # Enables SQLAlchemy model conversion

なぜモデルを分けるのか？

ProductCreate: クライアントはidを提供しません（データベースが生成します）
ProductUpdate: 部分的な更新のためにすべてのフィールドがオプション
ProductResponse: データベースからのidを含みます

6. APIエンドポイントを構築する

それでは、main.pyに完全なAPIを構築しましょう：

python

import logging
from fastapi import FastAPI, Depends, HTTPException, status
from models import ProductCreate, ProductUpdate, ProductResponse
import database_models
from database import session, engine
from sqlalchemy.orm import Session

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="Products API",
    description="A simple FastAPI backend for managing products",
    version="1.0.0"
)

# Create database tables
database_models.Base.metadata.create_all(bind=engine)

# Dependency: Database session management
def get_db():
    db = session()
    try:
        yield db
    finally:
        db.close()

💡 実際のプロジェクトでは、データベースマイグレーションにAlembicの使用を検討してください。

6.1. GET /products - すべての製品を取得

python

@app.get("/products", response_model=list[ProductResponse], tags=["Products"])
def get_products(db: Session = Depends(get_db)):
    try:
        logger.info("Fetching all products")
        products = db.query(database_models.Product).all()
        return products
    except Exception as e:
        logger.error(f"Error fetching products: {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to retrieve products"
        )

重要な概念：

response_model=list[ProductResponse]: レスポンス構造を検証
Depends(get_db): データベースセッションを自動的に注入
HTTPException: 適切なHTTPエラーレスポンスを返す

6.2. GET /products/:id - 単一製品を取得

python

@app.get("/products/{product_id}", response_model=ProductResponse, tags=["Products"])
def get_product(product_id: int, db: Session = Depends(get_db)):
    logger.info(f"Fetching product with ID: {product_id}")

    product = db.query(database_models.Product).filter(
        database_models.Product.id == product_id
    ).first()

    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {product_id} not found"
        )

    return product

6.3. POST /products - 新しい製品を作成

python

@app.post(
    "/products",
    response_model=ProductResponse,
    status_code=status.HTTP_201_CREATED,
    tags=["Products"]
)
def add_product(product: ProductCreate, db: Session = Depends(get_db)):
    try:
        logger.info(f"Creating new product: {product.name}")

        # Pydantic validates the request automatically
        db_product = database_models.Product(**product.model_dump())
        db.add(db_product)
        db.commit()
        db.refresh(db_product)  # Get the generated ID

        logger.info(f"Product created with ID: {db_product.id}")
        return db_product
    except Exception as e:
        logger.error(f"Error creating product: {e}")
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to create product"
        )

何が起きているのか？

FastAPIは自動的にProductCreateに対してリクエストボディを検証します
無効なデータ（負の価格、空の名前）は422エラーを返します
db.refresh()はデータベースから自動生成されたIDを取得します

6.4. PUT /products/:id - 製品を更新

python

@app.put("/products/{product_id}", response_model=ProductResponse, tags=["Products"])
def update_product(product_id: int, product: ProductUpdate, db: Session = Depends(get_db)):
    logger.info(f"Updating product with ID: {product_id}")

    db_product = db.query(database_models.Product).filter(
        database_models.Product.id == product_id
    ).first()

    if not db_product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {product_id} not found"
        )

    try:
        # Update only fields that were provided
        update_data = product.model_dump(exclude_unset=True)
        for field, value in update_data.items():
            setattr(db_product, field, value)

        db.commit()
        db.refresh(db_product)
        return db_product
    except Exception as e:
        logger.error(f"Error updating product: {e}")
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to update product"
        )

💡 exclude_unset=Trueは実際にリクエストで提供されたフィールドのみを含めるため、部分的な更新が可能になります。

6.5. DELETE /products/:id - 製品を削除

python

@app.delete("/products/{product_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Products"])
def delete_product(product_id: int, db: Session = Depends(get_db)):
    logger.info(f"Deleting product with ID: {product_id}")

    product = db.query(database_models.Product).filter(
        database_models.Product.id == product_id
    ).first()

    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {product_id} not found"
        )

    try:
        db.delete(product)
        db.commit()
        logger.info(f"Product deleted successfully")
        return None  # 204 No Content
    except Exception as e:
        logger.error(f"Error deleting product: {e}")
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to delete product"
        )

7. 便利なMakefile 🛠

一般的なタスクを簡素化するためのMakefileを作成します：

makefile

.PHONY: help install run db-up db-down clean setup env

help:
	@echo "Available commands:"
	@echo "  make install   - Install dependencies"
	@echo "  make run       - Run the API server"
	@echo "  make db-up     - Start PostgreSQL"
	@echo "  make db-down   - Stop PostgreSQL"
	@echo "  make setup     - Complete setup"

install:
	pip install -r requirements.txt

run:
	uvicorn main:app --reload

db-up:
	docker-compose up -d
	@echo "Database is ready!"

db-down:
	docker-compose down

setup:
	@make install
	@make db-up
	@echo "Setup complete! Run 'make run' to start."

env:
	@if [ ! -f .env ]; then \
		echo "DATABASE_URL=postgresql://admin:admin123@localhost:5432/products_db" > .env; \
		echo ".env file created"; \
	fi

これで、シンプルなコマンドが使用できます：

bash

~/products-api $ make setup  # プロジェクトをセットアップ
~/products-api $ make run    # APIを開始

8. APIドキュメント 🔨

http://localhost:8000/docs または http://localhost:8000/redoc にアクセスして、自動生成されたAPIドキュメントを確認してください。

以下を試してみてください：

製品を作成：

json

POST /products
{
  "name": "Laptop",
  "description": "MacBook Pro",
  "price": 1999.99,
  "quantity": 10
}

すべての製品を取得：

GET /products

製品を更新：

json

PUT /products/1
{
  "price": 1799.99
}

製品を削除：

DELETE /products/1

🔗 完全なコードはGitHubで確認できます

まとめ

FastAPIは自動検証、ドキュメントの組み込みサポート、型安全性を標準で提供します
作成、更新、レスポンス操作には別々のPydanticモデルを使用します
SQLAlchemy ORMはデータベース操作を抽象化します
**Depends()**はデータベースセッションの依存性注入を可能にします
公式ドキュメントからFastAPIについて詳しく学ぶことができます

FastAPIについて ⚡️

FastAPI固有の機能~

APIを構築する 🚀

0. 前提条件とセットアップ

1. シンプルに始める

2. データベース設定

3. SQLAlchemyのセットアップ

4. データベースモデルを定義する

5. Pydanticモデルを定義する

6. APIエンドポイントを構築する

6.1. GET /products - すべての製品を取得

6.2. GET /products/:id - 単一製品を取得

6.3. POST /products - 新しい製品を作成

6.4. PUT /products/:id - 製品を更新

6.5. DELETE /products/:id - 製品を削除

7. 便利なMakefile 🛠

8. APIドキュメント 🔨

まとめ

`FastAPI`について ⚡️