如何避免执行Alembic命令后自定义Python logging日志器失效?
解决Alembic执行迁移后Python logging日志停止输出的问题
问题重现
在Python应用启动阶段调用Alembic执行数据库迁移后,自定义日志器后续无法输出日志,代码示例如下:
import logging import alembic.command from alembic.config import Config as AlembicConfig logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger("app") logger.debug("Applying alembic migrations.") alembic_config = AlembicConfig("alembic.ini") alembic_config.attributes["sqlalchemy.url"] = connection_string alembic.command.upgrade(alembic_config, "head", tag="from_app") logger.debug("Terminating app.")
预期输出:
DEBUG:app:Applying alembic migrations. INFO [alembic.runtime.migration] Context impl PostgresqlImpl. INFO [alembic.runtime.migration] Will assume transactional DDL. DEBUG:app:Terminating app.
实际输出缺失最后一条Terminating app.日志,且后续所有日志均无法输出。
问题原因
并非logging系统被破坏,而是Alembic的command.upgrade()等命令函数在执行完成后会调用sys.exit(),直接终止了整个Python进程,导致后续代码(包括日志输出)根本没有机会执行。
解决方案
方案一:捕获SystemExit异常阻止进程终止
Alembic执行成功时会抛出SystemExit(0),失败时抛出非0状态码的SystemExit,只需捕获该异常即可让程序继续执行后续逻辑:
import logging import sys import alembic.command from alembic.config import Config as AlembicConfig logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger("app") logger.debug("Applying alembic migrations.") alembic_config = AlembicConfig("alembic.ini") alembic_config.attributes["sqlalchemy.url"] = connection_string try: alembic.command.upgrade(alembic_config, "head", tag="from_app") except SystemExit as e: # 若迁移失败,可根据退出码处理错误 if e.code != 0: logger.error(f"Alembic迁移失败,退出码:{e.code}") # 如需终止程序,可重新抛出异常 # sys.exit(e.code) # 现在可以正常输出日志 logger.debug("Terminating app.")
方案二:直接调用Alembic底层API(避免触发sys.exit)
绕过alembic.command封装的函数,直接调用迁移核心逻辑,这样不会触发进程退出:
import logging from sqlalchemy import create_engine from alembic.config import Config as AlembicConfig from alembic import script from alembic.runtime.migration import MigrationContext logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger("app") logger.debug("Applying alembic migrations.") alembic_config = AlembicConfig("alembic.ini") alembic_config.set_main_option("sqlalchemy.url", connection_string) # 创建数据库引擎 engine = create_engine(connection_string) # 获取脚本目录 script_dir = script.ScriptDirectory.from_config(alembic_config) # 执行迁移 with engine.connect() as conn: context = MigrationContext.configure( conn, script_directory=script_dir ) with context.begin_transaction(): context.run_migrations(tag="from_app") # 后续日志正常输出 logger.debug("Terminating app.")
验证效果
采用上述任一方案后,程序会完整执行所有逻辑,输出符合预期的日志内容。
内容的提问来源于stack exchange,提问作者Tim
相关产品推荐
相关产品推荐

