2025.10.14

Poetry × FastAPI × SQLAlchemy × Alembicで作る、モダンなPythonバックエンド環境構築ガイド

Table of Contents
Toggle

目的
ツールの役割
実装の手順
poetryプロジェクトの作成
SQLAlchemyの使用
DBモデルの作成
FastAPIアプリの作成
crud/user.pyの追加
routers/user.pyの追加
Fast APIアプリの起動
マイグレーションの設定
Alembicの初期化
alembic.iniの設定
migrations/env.pyの設定
マイグレーションの生成と実行
sqliteでテーブルの確認
主なalembicコマンドリスト
OpenAPIの確認

目的

FastAPIアプリの生成、依存管理、DBマイグレーションを、Poetryで一元管理することで、いままで複数のツールに依存していた作業を簡単にすることができます。

ツールの役割

ツール	役割	ひとことで言うと
Poetry	依存管理 & 仮想環境	Pythonのnpm的存在
FastAPI	Webアプリケーション	高速・型安全なAPIフレームワーク
SQLAlchemy	ORM	PythonからDBを操作
Alembic	DBマイグレーション	スキーマの変更を安全に管理

実装の手順

poetryプロジェクトの作成

poetry newコマンドによるプロジェクトを生成

poetry new fastapi-sample
cd fastapi-sample

poetry addコマンドにより、pythonパッケージをインストール

poetry add fastapi uvicorn sqlalchemy psycopg2 alembic

poetryプロジェクトの初期状態のフォルダ構成

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/

SQLAlchemyの使用

database.py でDBエンジン、セッション、Baseクラスの作成

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()

DBモデルの作成

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)

Baseクラスとモデルクラスを一括でインポートする準備

from ..database import Base
from .user import User

FastAPIアプリの作成

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!"}

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_user

routers/user.pyの追加

fastapiのAPIRouterによりメソッドをグループにまとめる

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

Fast APIアプリの起動

poetry runコマンドにより開発サーバーとFastAPIアプリを起動

poetry run uvicorn src.fastapi_sample.main:app --reload --port 8001

poetry run・・・Poetry の仮想環境内でコマンドを実行する
uvicorn・・・ASGIサーバー（FastAPIを動かすためのサーバー）を起動
src.fastapi_sample.main:app・・・実行するアプリの場所を指定
–reload・・・コード変更を検知して自動でサーバーを再起動（開発用）
–port 8001・・・サーバーをポート番号8001で起動する（デフォルト8000を変更）

http://localhost:8001 が表示されれば成功

マイグレーションの設定

Alembicの初期化

poetry run alembicコマンドによりマイグレーションを開始

poetry run alembic init migrations

migrationファイルが生成される

migrations/
├── env.py
├── script.py.mako
└── versions/
alembic.ini

alembic.iniの設定

接続URLを設定する

sqlalchemy.url = sqlite:///./test.db

migrations/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 ...  done

sqliteでテーブルの確認

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);

主なalembicコマンドリスト

Alembic設定フォルダの作成（例 migrationsフォルダ）
% poetry run alembic init migrations

新しいマイグレーションファイルを作成（空ファイル）
% poetry run alembic revision -m "message"

SQLAlchemy のモデル変更を検出してマイグレーションファイルを生成
% poetry run alembic revision --autogenerate -m "message"

1つ前のバージョンに戻す（＝直前の変更を取り消す）
% poetry run alembic downgrade -1

マイグレーション履歴一覧
% poetry run alembic history
<base> -> a86f66c6b53d, create user table

現在のリビジョンの確認
% poetry run alembic current
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
a86f66c6b53d

DBとモデルの差分確認
% poetry run alembic check
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
ERROR [alembic.util.messaging] Target database is not up to date.
  FAILED: Target database is not up to date.

リビジョンの確認
% poetry run alembic show a86f66c6b53d
Rev: a86f66c6b53d
Parent: <base>
Path: xxxxx/migrations/versions/2025_10_06_1652-a86f66c6b53d_create_user_table.py

    create user table
    
    Revision ID: a86f66c6b53d
    Revises:
    Create Date: 2025-10-06 16:52:33.741230

OpenAPIの確認

FastAPIではOpenAPI等のドキュメントが自動で生成される
http://localhost:8001/docs を開く

投稿者: s3lab
その他

Poetryとは？ – Python開発を劇的に快適にする依存管理ツール前の記事

OpenAI Codex – AIエンジニアに仕事を頼める次の記事