目的
Python製WebAPIを、依存管理からDBマイグレーションまで完全自動化し、手動の
venvやrequirements.txt管理から卒業する手法を紹介します。これらを組み合わせることで：
- 環境構築が自動化
- コードからDB設計を自動反映
- チーム開発でも「環境が一致」
という理想的な開発体験を実現します。
役割
| ツール | 役割 | ひとことで言うと |
|---|---|---|
| Poetry | 依存管理 & 仮想環境 | Pythonのnpm的存在 |
| FastAPI | Webアプリケーション | 高速・型安全なAPIフレームワーク |
| SQLAlchemy | ORM | PythonからDBを操作 |
| Alembic | DBマイグレーション | スキーマの変更を安全に管理 |
手順
プロジェクトの作成
poetry new fastapi-sample
cd fastapi-sampleパッケージのインストール
poetry add fastapi uvicorn sqlalchemy psycopg2 alembic初期状態のフォルダ構成
fastapi-sample/
├── pyproject.toml
├── README.md
├── src/
│ └── fastapi_sample/
│ └── __init__.py
└── tests/フォルダ構成を準備
mkdir -p src/fastapi_sample/{routers,models}
touch src/fastapi_sample/{main.py,database.py,models/__init__.py}src/
└── fastapi_sample/
├── __init__.py
├── main.py
├── database.py
├── models/
│ ├── __init__.py
│ └── user.py
└── routers/database.py の作成（SQLAlchemy設定）
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")
engine = create_engine(
DATABASE_URL,
connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
モデルの作成
from sqlalchemy import Column, Integer, String
from ..database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50))
email = Column(String(100), unique=True, index=True)
from ..database import Base
from .user import UserFastAPIアプリの作成
from fastapi import FastAPI
from . import models
from .database import engine
from src.fastapi_sample.routers import user as user_router
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
app.include_router(user_router.router)
@app.get("/")
def read_root():
return {"message": "Hello from FastAPI with src layout!"}
起動確認
poetry run uvicorn src.fastapi_sample.main:app --reload --port 8001http://localhost:8001 が表示されれば成功
Alembicの初期化
poetry run alembic init migrations以下が作成される
migrations/
├── env.py
├── script.py.mako
└── versions/
alembic.ini
alembic.iniの設定
接続URLを設定する
sqlalchemy.url = sqlite:///./test.dbmigrations/env.pyの設定
src をPythonパスに追加して、Baseを読み込むように編集
import sys
import pathlib
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
# --- 追加 ---
from src.fastapi_sample.database import Base
# -- Alembic設定 --
config = context.config
fileConfig(config.config_file_name)
target_metadata = Base.metadata
マイグレーションの生成と実行
% poetry run alembic revision --autogenerate -m "create users table"
Generating xxxx/fastapi-sample/migrations/versions/951446c9e367_create_users_table.py ... donesqliteでテーブルの確認
sqlite3 test.db
sqlite> .schema
CREATE TABLE users (
id INTEGER NOT NULL,
name VARCHAR(50),
email VARCHAR(100),
PRIMARY KEY (id)
);
CREATE INDEX ix_users_id ON users (id);
CREATE UNIQUE INDEX ix_users_email ON users (email);
crud/user.pyの追加
from sqlalchemy.orm import Session
from fastapi_sample.models.user import User
def get_user(db: Session, user_id: int):
return db.query(User).filter(User.id == user_id).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
return db.query(User).offset(skip).limit(limit).all()
def create_user(db: Session, name: str, email: str):
db_user = User(name=name, email=email)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def update_user(db: Session, user_id: int, name: str, email: str):
db_user = db.query(User).filter(User.id == user_id).first()
if not db_user:
return None
db_user.name = name
db_user.email = email
db.commit()
db.refresh(db_user)
return db_user
def delete_user(db: Session, user_id: int):
db_user = db.query(User).filter(User.id == user_id).first()
if not db_user:
return None
db.delete(db_user)
db.commit()
return db_userrouters/user.pyの追加
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from fastapi_sample.database import SessionLocal
from fastapi_sample.crud import user as crud_user
router = APIRouter(prefix="/users", tags=["Users"])
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@router.post("/")
def create_user(name: str, email: str, db: Session = Depends(get_db)):
return crud_user.create_user(db, name, email)
@router.get("/")
def read_users(db: Session = Depends(get_db)):
return crud_user.get_users(db)
@router.get("/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
user = crud_user.get_user(db, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.put("/{user_id}")
def update_user(user_id: int, name: str, email: str, db: Session = Depends(get_db)):
user = crud_user.update_user(db, user_id, name, email)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.delete("/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
user = crud_user.delete_user(db, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
OpenAPIの確認
FastAPIではOpenAPI等のドキュメントが自動で生成される
http://localhost:8001/docs を開く
