🦓

[FastAPI]DBへの接続とマイグレーション

2024/09/16に公開

PostgreSQL

はじめに

前回の記事の続きとして、FastAPIのDB接続とマイグレーション関連をまとめていきます。

環境構築

公式を参考しながら進めていきます。

必要なパッケージ

FastAPI自体にはマイグレーションやデータベース操作の機能が組み込まれていないため、外部パッケージを使用するのが一般的です。

・FastAPI
・PostgreSQL（Dockerコンテナ）
・Uvicorn（ASGIサーバ）
・SQLAlchemy（Pythonの代表的なORM）
・Alembic（SQLAlchemyと連携して動作するマイグレーションツール）
・psycopg2-binary (PythonのPostgreSQLデータベースアダプタ)
・python-dotenv（環境変数を管理するライブラリ）

GUIからデータベースへの接続を確認したい場合こちらのツールも必要です。
・DBeaver（DBクライアントツール)

.devcontainer/docker-compose.ymlに記載した接続情報でpostgresというデータベースに接続できます。この時点では空のデータベースです。

tl;dr

database.pyにDBの接続情報を追加する
SQLAlchemyを使ってモデルファイルを作成する
Alembicを使ってマイグレーションファイルを作成する

ディレクトリ構造はこのようになります。

.
├── app
│   ├── __init__.py
│   ├── __pycache__
│   ├── alembic
│   │   ├── README
│   │   ├── env.py
│   │   ├── script.py.mako
│   │   └── versions
│   │       └── f705942eb49e_create_user_table.py
│   ├── alembic.ini
│   ├── database.py
│   ├── main.py
│   └── models
│       └── user.py
└── requirements.txt

`app/database.py`を作成する

.envを作成します。

.env

DATABASE_TYPE=postgresql
DATABASE_HOST=postgres
DATABASE_PORT=5432
DATABASE_NAME=
DATABASE_USER=
DATABASE_PASSWORD=

DBへの接続情報を追加します。

app/database.py

import os
from dotenv import load_dotenv
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool

load_dotenv()

# 環境変数からデータベース接続情報を取得
DB_TYPE=os.getenv("DB_TYPE", "postgresql")
DB_USER = os.getenv("DB_USER", "postgres")
DB_PASSWORD = os.getenv("DB_PASSWORD", "postgres")
DB_HOST = os.getenv("DB_HOST", "postgres")
DB_PORT = os.getenv("DB_PORT", "5432")
DB_NAME = os.getenv("DB_NAME", "postgres")

# データベースURLを構築
DATABASE_URL = f"{DB_TYPE}://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"

# SQLAlchemyエンジンを作成
Engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    max_overflow=10,
    pool_size=20,
    pool_timeout=30,
    pool_recycle=1800,
)

# SessionLocalクラスを作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=Engine)

# 宣言的モデルのためのBaseクラスを作成
Base = declarative_base()

# データベースセッションを取得するための依存関数
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

ORM

ORMはオブジェクト・リレーショナル・マッピングの略である、FastAPIからデータベースへ接続するためのツールです。
ORMでは通常、SQLデータベースのテーブル(usersテーブル)を表すクラス(Class Userユーザークラス)を作成し、クラスの各属性は名前と型を持つカラムを表します。
ORMは、コード内のクラスとデータベースのテーブルを変換（マップ）するツールです。
そのクラスの各インスタンス・オブジェクト（user = User()）は、データベースの一レコードを表します。
そしてORMは、インスタンス・オブジェクトからDBへアクセスしようとしたときに、対応するテーブルから情報を取得するための作業を行います。

Pythonの一般的なORM：
・Django-ORM (Django フレームワークの一部)
・SQLAlchemy ORM (SQLAlchemy の一部、フレームワークとは独立)
・Peewee (フレームワークとは独立)

今回のアプリにおいて、SQLAlchemyを使用します。
https://www.sqlalchemy.org/

SQLAlchemyを使ってモデルファイルを作成する

まず、models/user.pyファイルでSQLAlchemyモデルを定義します。
今回はUserモデルを定義します。

models/user.py

import datetime
from sqlalchemy import Column, DateTime, Integer, String, Text, func
from app.database import Base
from pydantic import BaseModel


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True, nullable=False)
    username = Column(Text, nullable=False)
    email = Column(String(319), nullable=False, unique=True)
    created_at = Column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    updated_at = Column(
        DateTime(timezone=True),
        nullable=False,
        server_default=func.now(),
        onupdate=func.now(),
    )


class User(BaseModel):
    id: int
    username: str
    email: str
    created_at: datetime.datetime
    updated_at: datetime.datetime

    class Config:
        orm_mode = True

# 他のモデルとのリレーションも定義できます

Alembicを使ってマイグレーションファイルを作成する

次に、マイグレーションファイルの作成とデータベースへの反映を行います。
SQLAlchemyを使用している場合、通常はAlembicを使ってマイグレーションを管理します。

Alembicを初期化する

root@e3a333aa1237:/workspace/backend/app# alembic init alembic
  Creating directory '/workspace/backend/app/alembic' ...  done
  Creating directory '/workspace/backend/app/alembic/versions' ...  done
  Generating /workspace/backend/app/alembic/README ...  done
  Generating /workspace/backend/app/alembic/env.py ...  done
  Generating /workspace/backend/app/alembic.ini ...  done
  Generating /workspace/backend/app/alembic/script.py.mako ...  done
  Please edit configuration/connection/logging settings in '/workspace/backend/app/alembic.ini' before proceeding.

これにより、alembicディレクトリとalembic.iniファイルが作成されます。

`alembic.ini`ファイルを編集し、データベースのURLを設定する

# alembic.ini

# 他の設定は省略...

sqlalchemy.url = %(DB_TYPE)://%(DB_USER)s:%(DB_PASSWORD)s@%(DB_HOST)s:%(DB_PORT)s/%(DB_NAME)s

`alembic/env.py`ファイルを編集して、SQLAlchemyのモデルを読み込み

alembic/env.py

from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context

# この行を追加
from app.database import Base, DATABASE_URL

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# この行を追加
config.set_main_option("sqlalchemy.url", DATABASE_URL)

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = Base.metadata  # この行を変更

# 他の部分は省略...

マイグレーションファイルを作成する

root@e3a333aa1237:/workspace/backend/app# alembic revision -m "create user table"
  Generating /workspace/backend/app/alembic/versions/f705942eb49e_create_user_table.py ...  done

これにより、alembic/versions/ディレクトリに新しいマイグレーションファイルが作成されます。

初期のapp/alembic/versions/xxx_create_user_table.py

app/alembic/versions/xxx_create_user_table.py

"""create user table

Revision ID: f705942eb49e
Revises: 
Create Date: 2024-09-16 12:40:34.653725

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = 'f705942eb49e'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
   pass


def downgrade() -> None:
   pass

userテーブルに追加するカラムを定義します。

app/alembic/versions/xxx_create_user_table.py

"""create user table

Revision ID: f705942eb49e
Revises: 
Create Date: 2024-09-16 12:40:34.653725

"""

import datetime
from time import timezone
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = "f705942eb49e"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
   op.create_table(
       "users",
       sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
       sa.Column("username", sa.String(50), nullable=False, comment="ユーザー名"),
       sa.Column(
           "email",
           sa.String(128),
           unique=True,
           nullable=False,
           comment="メールアドレス",
       ),
       sa.Column(
           "created_at",
           sa.DateTime(timezone=True),
           nullable=False,
           default=datetime.datetime.now,
           comment="作成日時",
       ),
       sa.Column(
           "updated_at",
           sa.DateTime(timezone=True),
           nullable=False,
           onupdate=datetime.datetime.now,
           comment="更新日時",
       ),
   )


def downgrade() -> None:
   op.drop_table("users")

公式にあるチュートリアルもぜひ参考にしてみてください。

マイグレーションをデータベースに適用する

root@e3a333aa1237:/workspace/backend/app# alembic upgrade head
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> f705942eb49e, create user table

DBクライアントツールからもしくはpsqlコマンドでusersテーブルが作成されたことを確認します。

alembicコマンド

マイグレーション関連のalembicコマンドをまとめました。

マイグレーションの初期化:

alembic init alembic

新しいマイグレーションの作成:

alembic revision --autogenerate -m "説明文"

モデルの変更を検出し、新しいマイグレーションファイルを自動生成します。

最新のマイグレーションを適用:

alembic upgrade head

特定のバージョンにアップグレード:

alembic upgrade <revision>

1つ前のバージョンにダウングレード:

alembic downgrade -1

特定のバージョンにダウングレード:

alembic downgrade <revision>

現在のリビジョンを表示:

alembic current

データベースの現在のリビジョンを表示します。

マイグレーション履歴の表示:

alembic history

最初のリビジョンまでダウングレード:

alembic downgrade <最初のリビジョンID>

または、完全に初期状態に戻す場合:

alembic downgrade base

SQLの生成（適用はしない）:

alembic upgrade head --sql

マイグレーションを適用せずに、生成される SQL を表示します。

オフラインモードでのマイグレーション:

alembic upgrade head --sql > migration.sql

マイグレーションの SQL をファイルに出力します。

終わりに

FastAPIからPostgreSQLへの接続、
SQLAlchemyを使ったモデルの定義とAlembicを使ったデータベースのマイグレーションをまとめてみました。
誰かの参考になれば嬉しいです。