SQLAlchemy Table Migration with Alembic

supabase and fastapi Python

FastAPIにてSQLAlchemyを用いたテーブルマイグレーション方法に関する備忘録。

こちらの記事の続きで、Supabase(PostgreSQL)にテーブルの変更内容を反映させることを目的としています。

SQLAlchemy公式のマイグレーションツールであるAlembicを導入する手順を紹介します。

✅ Alembic 導入手順

1. 依存ライブラリを追加

requirements.txt に追加：

alembic==1.13.1

インストール：

pip install alembic

and add alembic==1.15.2 in requirements.txt

2.初期化

alembic init alembic

alembic folder and alembic.ini file will be created.

3.alembic.ini

sqlalchemy.url =

Not to specify database url here.

4.alembic/env.py

初期化で生成されたenv.pyを以下のように変更する。

import os
from logging.config import fileConfig

from dotenv import load_dotenv
from sqlalchemy import pool

from alembic import context

load_dotenv()
DATABASE_URL = os.getenv("SYNC_DATABASE_URL_ALEMBIC")

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

# 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 app.models.base import Base

# target_metadata = mymodel.Base.metadata
target_metadata = Base.metadata


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    if not url:
        url = DATABASE_URL

    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online() -> None:
    url = config.get_main_option("sqlalchemy.url")
    if not url:
        url = DATABASE_URL

    from sqlalchemy import create_engine
    connectable = create_engine(url, poolclass=pool.NullPool)

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

ポイント

  • FastAPI + Alembic の場合、SYNC(同期ドライバ)psycopg2を使う必要がある。
  • アプリ側で +asyncpg 、ASYNC(非同期)として接続。

.env

DATABASE_URL=postgresql+asyncpg://[YOUR_PROJECT].xxxx.supabase.com:5432/postgres # app
SYNC_DATABASE_URL_ALEMBIC=postgresql://[YOUR_PROJECT].xxxx.supabase.com:6543/postgres #Alembic

app/models/__init__.py

from .message import Message
from .user import User
from .conversation import Conversation

app/models/base.py

from sqlalchemy.orm import declarative_base

Base = declarative_base()

このBaseをalembic/env.py 内で使う

Modify Message Model

from sqlalchemy import Column, String, ForeignKey, DateTime, Enum
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from datetime import datetime

class Message(Base):
    __tablename__ = "messages"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    conversation_id = Column(UUID(as_uuid=True), ForeignKey("conversations.id"), nullable=False)

    role = Column(String, nullable=False)  # 'user' or 'assistant'
    content = Column(String, nullable=False)
    translated_content = Column(String, nullable=True)
    
    audio_url = Column(String, nullable=True)  # Add new column
    created_at = Column(DateTime, default=datetime.utcnow)

    conversation = relationship("Conversation", back_populates="messages")

5.Create migration file

alembic revision --autogenerate -m "add audio_url"

alembic/versions 内にファイルが自動生成される。

生成されたファイルがマイグレーションの変更内容になる。

たまに間違っていることがあるので、中身を適宜確認する。

以下のようなファイルが生成されれば、OK。

alembic/versions/32d973e73589_add_audio_url.py

"""add audio url 

Revision ID: 32d973e73589
Revises: 
Create Date: 2025-04-12 15:15:43.383714

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa

# revision identifiers, used by Alembic.
revision: str = '32d973e73589'
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:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('message', sa.Column('audio_url', sa.String(), nullable=True))
    # ### end Alembic commands ###

def downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column('message', 'audio_url')
    # ### end Alembic commands ###

6.Execute migration

alembic upgrade head

コマンド実行後、テーブル変更内容が反映されていればOK。

スポンサー広告