概要

Flask-SQLAlchemyを使ったDocker環境下のFlaskとDB(Postgres)との接続方法
Model作成後のマイグレーションファイルの作成およびマイグレーションの実行方法
アプリケーションを実行した際のSQLの閲覧方法

について解説します

前提

DBはPostgresを使用
Flaskのバージョン3を使用

Flask-SQLAlchemyを使ったアプリケーションとDBの接続設定

docker-compose.ymlにアプリケーションとDBの設定を記載します

今回はFlask-SQLAlchemyについての記事なので詳細は説明しませんが、気になる方は以下の記事を参考にしてみてください

docker-compose.yml

services:
  db:
    container_name: db
    build:
      context: .
      dockerfile: containers/postgres/Dockerfile
    platform: linux/x86_64
    volumes:
      - db_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"
    # ヘルスチェック
    healthcheck:
      test: pg_isready -U "${POSTGRES_USER:-postgres}" || exit 1
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 30s
    environment:
      - POSTGRES_NAME
      - POSTGRES_USER
      - POSTGRES_PASSWORD
  app:
    container_name: app
    build:
      context: .
      dockerfile: containers/flask/Dockerfile
    volumes:
      - ./application:/code
    ports:
      - "8000:8000"
      - "8080:8080"
    command: sh -c "/usr/local/bin/entrypoint.sh"
    env_file:
      - .env
    depends_on:
      db:
        condition: service_healthy
volumes:
  db_data:

.envファイルにDB用の設定などを記載します

.env

POSTGRES_NAME=postgres
POSTGRES_USER=postgres
POSTGRES_PASSWORD=postgres
DB_URL=postgresql://postgres:postgres@db:5432/postgres

DB_URLにアプリケーションがDBへ接続するときの接続情報を記載します
今回はPostgresを使用するので以下のように必要な情報を記載していきます

DB_URL=postgresql://{ユーザ名}:{パスワード}@{ホスト名(コンテナ名)}:{ポート番号}/{DB名}

詳細はSQLAlchemyのドキュメントに記載されています

main.pyに以下のように設定します
SQLALCHEMY_DATABASE_URIの環境変数に.envファイルのDB_URLを設定します

main.py

from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get("DB_URL")
db = SQLAlchemy()
# flask_sqlalchemyの初期化
db.init_app(app)

マイグレーションの設定

flask_migrateを使ってマイグレーションを行います

main.py

from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get("DB_URL")
db = SQLAlchemy()
db.init_app(app)
migrate = Migrate(app, db)
# flask_migrateの初期化
migrate.init_app(app, db)

以下のコマンドを使ってマイグレーション設定の初期化を行います

flask db init

上記のコマンドを実行すると、以下のようにmigrationsフォルダとそのファイル群が自動生成されます

└── application
    ├── .vscode
    │   └── launch.json
    ├── main.py
    └── migrations
        ├── README
        ├── alembic.ini
        ├── env.py
        ├── script.py.mako
        └── versions

  Creating directory '/code/migrations' ...  done
  Creating directory '/code/migrations/versions' ...  done
  Generating /code/migrations/alembic.ini ...  done
  Generating /code/migrations/script.py.mako ...  done
  Generating /code/migrations/env.py ...  done
  Generating /code/migrations/README ...  done
  Please edit configuration/connection/logging settings in '/code/migrations/alembic.ini' before proceeding.

マイグレーションを実行するためにUserのModelを作成してみます

main.py

class User(db.Model):
    __tablename__ = "users"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def to_dict(self):
        return {"id": self.id, "name": self.name, "email": self.email}

以下のコマンドを使ってマイグレーションファイルを作成します

flask db migrate -m "Initial migration."

以下のようにversionsの下にマイグレーションファイルが作成されたら成功です

INFO  [alembic.autogenerate.compare] Detected added table 'users'
INFO  [sqlalchemy.engine.Engine] COMMIT
  Generating /code/migrations/versions/838a24c05dea_initial_migration.py ...  done

マイグレーションファイルの中身は以下の通りです

"""Initial migration.

Revision ID: 838a24c05dea
Revises: 
Create Date: 2025-01-15 06:17:26.727794

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '838a24c05dea'
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('users',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.String(length=80), nullable=False),
    sa.Column('email', sa.String(length=120), nullable=False),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('email')
    )
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('users')
    # ### end Alembic commands ###

マイグレーションファイルを作成しただけでは実際にDBに反映されないので以下のコマンドでマイグレーションファイルの中身を反映させます

flask db upgrade

以下のようにDBにマイグレーションファイルの内容が適用されたら成功です

INFO  [sqlalchemy.engine.Engine] INSERT INTO alembic_version (version_num) VALUES ('838a24c05dea') RETURNING alembic_version.version_num
INFO  [sqlalchemy.engine.Engine] [generated in 0.00024s] {}
INFO  [sqlalchemy.engine.Engine] COMMIT

postgres=# \d users
                                    Table "public.users"
 Column |          Type          | Collation | Nullable |              Default              
--------+------------------------+-----------+----------+-----------------------------------
 id     | integer                |           | not null | nextval('users_id_seq'::regclass)
 name   | character varying(80)  |           | not null | 
 email  | character varying(120) |           | not null | 
Indexes:
    "users_pkey" PRIMARY KEY, btree (id)
    "users_email_key" UNIQUE CONSTRAINT, btree (email)

実行したSQLを見るには

クエリを実行するたびに見たい場合

SQLALCHEMY_ECHOをTrueにするとSQLAlchemyのORMを実行した際のSQLを閲覧できます

app.config['SQLALCHEMY_ECHO'] = True

今回はUserテーブルからデータを全て取得するクエリを実行します
以下のようにクエリのログが表示されたら成功です

>>> from main import User
>>> User.query.all()
2025-01-15 05:27:58,628 INFO sqlalchemy.engine.Engine select pg_catalog.version()
2025-01-15 05:27:58,628 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-01-15 05:27:58,630 INFO sqlalchemy.engine.Engine select current_schema()
2025-01-15 05:27:58,630 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-01-15 05:27:58,632 INFO sqlalchemy.engine.Engine show standard_conforming_strings
2025-01-15 05:27:58,633 INFO sqlalchemy.engine.Engine [raw sql] {}
2025-01-15 05:27:58,634 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2025-01-15 05:27:58,638 INFO sqlalchemy.engine.Engine SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email 
FROM users
2025-01-15 05:27:58,638 INFO sqlalchemy.engine.Engine [generated in 0.00042s] {}
[<User 1>]

これまで実行したSQLの履歴をまとめて閲覧したい場合

Flaskのバージョン3からget_debug_queriesではなく、get_recorded_queriesを使用することになります
(2以下を使用する場合はget_debug_queriesを使用することになります)

以下がrecord_queries.pyのソースコードです

flask_sql_alchemy.record_queries.py

def get_recorded_queries() -> list[_QueryInfo]:
    """Get the list of recorded query information for the current session. Queries are
    recorded if the config :data:`.SQLALCHEMY_RECORD_QUERIES` is enabled.

    Each query info object has the following attributes:

    ``statement``
        The string of SQL generated by SQLAlchemy with parameter placeholders.
    ``parameters``
        The parameters sent with the SQL statement.
    ``start_time`` / ``end_time``
        Timing info about when the query started execution and when the results where
        returned. Accuracy and value depends on the operating system.
    ``duration``
        The time the query took in seconds.
    ``location``
        A string description of where in your application code the query was executed.
        This may not be possible to calculate, and the format is not stable.

    .. versionchanged:: 3.0
        Renamed from ``get_debug_queries``.

    .. versionchanged:: 3.0
        The info object is a dataclass instead of a tuple.

    .. versionchanged:: 3.0
        The info object attribute ``context`` is renamed to ``location``.

    .. versionchanged:: 3.0
        Not enabled automatically in debug or testing mode.
    """
    return g.get("_sqlalchemy_queries", [])  # type: ignore[no-any-return]

SQLALCHEMY_RECORD_QUERIESをTrueに設定します

main.py

app.config["SQLALCHEMY_RECORD_QUERIES"] = True

シェルを実行後、flask_sqlalchemy.record_queriesからget_recorded_queriesをimportし、実行すると以下のようにこれまで実行されたSQLの履歴が全て取得できれば成功です

>>> from flask_sqlalchemy.record_queries import get_recorded_queries
>>> get_recorded_queries()
[_QueryInfo(statement='SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email \nFROM users', parameters=[{}], start_time=1385.069575485, end_time=1385.071073211, location='<unknown>'), _QueryInfo(statement='SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email \nFROM users \nWHERE users.id = %(pk_1)s', parameters=[{'pk_1': 3}], start_time=1417.486363535, end_time=1417.487912003, location='<unknown>')]
>>> User.query.all()
2025-01-15 05:29:13,911 INFO sqlalchemy.engine.Engine SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email 
FROM users
2025-01-15 05:29:13,912 INFO sqlalchemy.engine.Engine [cached since 75.27s ago] {}
[<User 1>]
>>> get_recorded_queries()
[_QueryInfo(statement='SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email \nFROM users', parameters=[{}], start_time=1385.069575485, end_time=1385.071073211, location='<unknown>'), _QueryInfo(statement='SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email \nFROM users \nWHERE users.id = %(pk_1)s', parameters=[{'pk_1': 3}], start_time=1417.486363535, end_time=1417.487912003, location='<unknown>'), _QueryInfo(statement='SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email \nFROM users', parameters=[{}], start_time=1460.341334045, end_time=1460.342446944, location='<unknown>')]

参考