概要
- 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>')]
参考