Docker化Python项目:实现应用启动时自动执行dbmate数据库迁移
方案一:将迁移逻辑嵌入应用启动流程
1. 轻量迁移实现(无需第三方CLI工具)
不用依赖dbmate、alembic这类CLI,直接把迁移逻辑整合到应用代码中,用自定义的迁移表跟踪执行状态。
首先创建migrations模块存放迁移脚本:
# migrations/__init__.py from typing import List, Callable import psycopg2 # 替换为你使用的数据库驱动 from myapp.config import DB_URL # 按时间顺序定义迁移脚本,每个脚本包含唯一ID和执行函数 MIGRATIONS: List[tuple[str, Callable]] = [ ( "20240520_create_users_table", lambda cur: cur.execute(""" CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, username VARCHAR(50) UNIQUE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) ), # 后续迁移脚本依次添加 ] def run_migrations(): # 先初始化迁移记录表 with psycopg2.connect(DB_URL) as conn: with conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS migration_history ( migration_id VARCHAR(50) PRIMARY KEY, executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() # 查询已执行的迁移 with conn.cursor() as cur: cur.execute("SELECT migration_id FROM migration_history") executed_ids = {row[0] for row in cur.fetchall()} # 执行未完成的迁移 for migrate_id, migrate_func in MIGRATIONS: if migrate_id not in executed_ids: with conn.cursor() as cur: migrate_func(cur) conn.commit() # 记录迁移完成状态 with conn.cursor() as cur: cur.execute("INSERT INTO migration_history (migration_id) VALUES (%s)", (migrate_id,)) conn.commit() print(f"Completed migration: {migrate_id}")
然后在应用启动入口调用迁移函数:
# main.py from myapp import app from migrations import run_migrations if __name__ == "__main__": run_migrations() app.run(host="0.0.0.0", port=8000)
这种方式的迁移逻辑和应用代码完全绑定,打包进精简venv也不会有问题,不需要额外安装CLI工具。
2. Docker镜像适配
Dockerfile只需保证依赖安装完成,启动命令直接运行应用入口即可:
FROM python:3.11-slim WORKDIR /app # 创建精简venv RUN python -m venv /opt/venv ENV PATH="/opt/venv/bin:$PATH" # 安装运行时依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码和迁移模块 COPY myapp/ ./myapp/ COPY migrations/ ./migrations/ COPY main.py . # 启动时自动执行迁移+启动应用 CMD ["python", "main.py"]
3. 适配testcontainers测试
测试时启动PostgreSQL容器后,直接调用run_migrations()即可,无需额外CLI操作:
# tests/conftest.py import pytest from testcontainers.postgres import PostgresContainer from myapp.config import set_db_url from migrations import run_migrations @pytest.fixture(scope="session") def db_container(): with PostgresContainer("postgres:15") as container: db_url = container.get_connection_url() set_db_url(db_url) run_migrations() yield container
方案二:常规独立迁移流程(复杂场景首选)
如果自定义迁移逻辑无法满足复杂需求,或者团队更习惯成熟工具,常规的独立迁移流程更合适:
1. 部署流程
- 用alembic或dbmate作为迁移工具,单独构建包含迁移工具和脚本的镜像(或在部署流水线中安装工具)
- 部署时先执行迁移,再启动应用:
- 运行迁移命令:
docker run myapp-migrate alembic upgrade head(或流水线中直接调用CLI) - 启动应用容器
- 运行迁移命令:
2. 测试适配
测试时通过调用CLI工具执行迁移:
# tests/conftest.py import pytest import subprocess from testcontainers.postgres import PostgresContainer from myapp.config import set_db_url @pytest.fixture(scope="session") def db_container(): with PostgresContainer("postgres:15") as container: db_url = container.get_connection_url() set_db_url(db_url) # 调用alembic执行迁移 subprocess.run(["alembic", "upgrade", "head"], env={"DATABASE_URL": db_url}, check=True) yield container
独立迁移流程的优势
- 成熟工具支持复杂操作:自动生成迁移脚本、回滚、分支迁移等,自定义方案难以覆盖
- 权限分离:迁移用高权限账号,应用运行用低权限账号,更安全
- 可追溯性:工具维护的迁移记录更规范,团队协作时更容易跟踪状态
总结
如果你的应用迁移逻辑简单,追求启动时自动执行的便利性,方案一完全适配你的需求;如果应用有复杂迁移场景,或需要更规范的团队协作流程,方案二的独立迁移流程更实用——虽然测试时多一步CLI调用,但成熟工具带来的稳定性和功能扩展性远大于这点复杂度。
内容的提问来源于stack exchange,提问作者caeus
相关产品推荐
相关产品推荐

