FastAPIでSQL/マイグレーションを管理するために利用
まずは素のpythonで試す

Python 3.11.7
SQLAlchemy 2.0.25
alembic 1.13.1

not75743

sqlalchemy/alembicについて

SQLAlchemy

チュートリアル

チュートリアルはここ

できること

SQLコマンドの生成
データベースとのやり取り
トランザクションの管理

alembic

チュートリアル

チュートリアルはここ

できること

マイグレーションスクリプトの生成
スキーマのバージョン管理
スキーマ変更の実行とロールバック

install

pip install sqlalchemy
pip install psycopg2-binary

not75743

ファイル群

このあたりを参考に

構成

.
├── database.py
├── main.py
├── models.py
└── operation.py

database.py

データベースエンジンとセッションの設定

database.py

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(
    "postgresql://username:password@postgres:5432/testdb", echo=False
)
Session = sessionmaker(bind=engine)

models.py

データベーステーブルに対応するモデルクラスを設定

models.py

# models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

from database import engine

Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)
    password = Column(String)

    def __repr__(self):
        return "<User('%s','%s', '%s')>" % (self.name, self.age, self.password)


Base.metadata.create_all(engine)

operation.py

ユーザーの作成、更新、削除などの機能

operation.py

# user_management.py
from database import Session
from models import User


def add_user(name, age, password):
    session = Session()
    new_user = User(name=name, age=age, password=password)
    session.add(new_user)
    session.commit()
    return new_user


def get_user_by_id(user_id):
    session = Session()
    return session.query(User).filter_by(id=user_id).first()


def get_all_users():
    session = Session()
    return session.query(User).all()


def update_user_by_id(user_id, new_name, new_age, new_password):
    session = Session()
    user = session.query(User).filter_by(id=user_id).first()
    if user:
        user.name = new_name
        user.age = new_age
        user.password = new_password
        session.commit()
        return user
    return None


def delete_user_by_id(user_id):
    session = Session()
    user = session.query(User).filter_by(id=user_id).first()
    if user:
        session.delete(user)
        session.commit()
        return True
    return False


def delete_all_users():
    session = Session()
    session.query(User).delete()
    session.commit()
    return True

main.py

operationモジュールを呼び出して各種操作を行う
必要に応じてコメントアウト

main.py

# main.py
from operation import (
    add_user,
    delete_all_users,
    delete_user_by_id,
    get_all_users,
    get_user_by_id,
    update_user_by_id,
)

# ユーザの追加
user = add_user("example1", 20, "example1password")

# IDでユーザ検索
user = get_user_by_id(35)
print(user)

# 全ユーザ検索
all_users = get_all_users()
print(all_users)

# IDでユーザ更新
update_user_by_id(35, "example1_updated", 21, "newpassword")

# IDでユーザ削除
delete_user_by_id(35)

# 全ユーザ削除
# delete_all_users()

not75743

Sessionを使ったデータベース操作

このあたりを参考に

not75743

Sessionのaddやqueryのドキュメント

必要なものがあったらここを見る

not75743

alembicによるカラム名の変更

userからusernameに変更する

1. コードの変更

models.py

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
-   name = Column(String)
+   username = Column(String)
    age = Column(Integer)
    password = Column(String)

2. （必要に応じて）alembicの初期化

alembicディレクトリができることを確認する

alembic init alembic

3. Alembicの設定ファイルの編集

データベースの接続情報を設定

alembic/alembic.ini

sqlalchemy.url = postgresql://username:password@postgres:5432/testdb

4. 環境設定スクリプトの編集

Userモデルの変更を検知できるようにインポートする
Userモデルのインポートが必要

alembic/env.py

from models import Base
target_metadata = Base.metadata

5. マイグレーションスクリプトの生成

マイグレーションスクリプトを作成する

alembic revision --autogenerate -m "Change name to username in User model"

6. マイグレーションの適用

alembic upgrade head

not75743

alembicによるカラム名の変更②

usernameからuserに戻す

1. コードの変更

models.py

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
-   username = Column(String)
+   name = Column(String)
    age = Column(Integer)
    password = Column(String)

2. マイグレーションスクリプトの生成

マイグレーションスクリプトを作成する

alembic revision --autogenerate -m "Change name to user in User model"

3. マイグレーションの適用

alembic upgrade head

not75743

alembicのマイグレーションスクリプト

almbic/versions/xxxxx.pyの形で存在します。
--autogenerate実行毎に生成されるっぽい
ひつように応じてこちらもレビューを行う

$ tree -I "__pycache__" alembic
alembic
├── README
├── env.py
├── script.py.mako
└── versions
    ├── 3cda0f6b8ed8_change_name_to_username_in_user_model.py
    └── bf30ccaaf31e_change_username_back_to_name_in_user_.py