升级SQLAlchemy至2.0后Flask-Migrate执行db upgrade时认证失败

• 将Flask应用的SQLAlchemy从1.4.46升级至2.0.1后，执行flask db upgrade（Flask-Migrate）时触发PostgreSQL密码认证失败错误
• 正常启动并运行Flask应用时，数据库连接完全正常
• 升级前（1.4.46版本）或降级回1.4.46后，flask db upgrade操作均能正常执行

相关配置 #

数据库URI构造代码 #

SQLALCHEMY_DATABASE_URI = f"postgresql+psycopg2://{PG_USER}:{PG_PASSWORD}@{PG_HOST}:5432/{PG_DB}?{urlencode(LIBPQ_PARAMS)}"

生成的URI示例 #

postgresql+psycopg2://user:xxx@example.com:5432/exampledb?connect_timeout=10&keepalives=1&keepalives_idle=60&keepalives_interval=10&keepalives_count=5&sslmode=require

错误日志 #

PostgreSQL服务端日志 #

PG-00000 LOG: connection received: host=10.101.15.236 port=53150
PG-28P01 FATAL: password authentication failed for user "user"
PG-28P01 DETAIL: Connection matched pg_hba.conf line 18: "hostssl all all 0.0.0.0/0 scram-sha-256"

Python栈追踪 #

File "/home/so/venv/lib64/python3.8/site-packages/psycopg2/__init__.py", line 122, in connect
    conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) connection to server at "example.com" (10.101.1.28), port 5432 failed: FATAL:  password authentication failed for user "user"

1. 修正密码URL编码逻辑 #

SQLAlchemy 2.0对URI解析逻辑做了调整，1.4版本可能自动兼容未编码的特殊字符密码，但2.0版本需要手动确保密码已正确URL编码。修改URI构造代码，单独对密码进行编码：

from urllib.parse import quote_plus, urlencode

SQLALCHEMY_DATABASE_URI = f"postgresql+psycopg2://{PG_USER}:{quote_plus(PG_PASSWORD)}@{PG_HOST}:5432/{PG_DB}?{urlencode(LIBPQ_PARAMS)}"

2. 验证Flask-Migrate的配置加载 #

Flask-Migrate执行命令时可能未加载到与正常运行时一致的配置：

• 确认执行flask db upgrade时，环境变量（如PG_PASSWORD）已正确设置
• 检查migrate.init_app(app, db)是否正确关联了应用的配置实例
• 可在迁移脚本中临时添加打印逻辑，对比current_app.config['SQLALCHEMY_DATABASE_URI']与正常运行时的URI是否一致

3. 检查SCRAM认证兼容性 #

PostgreSQL日志显示使用SCRAM-SHA-256认证，需确保客户端兼容：

• 升级psycopg2到2.8及以上版本（该版本开始支持SCRAM认证）
• 若使用psycopg2-binary，确认版本与SQLAlchemy 2.0兼容
• 可尝试在URI中显式添加password_encryption=scram-sha-256参数，或检查数据库用户的加密方式与客户端匹配

4. 直接测试SQLAlchemy 2.0连接 #

编写简单脚本跳过Flask-Migrate，验证SQLAlchemy 2.0本身能否连接数据库：

from sqlalchemy import create_engine

uri = "postgresql+psycopg2://user:xxx@example.com:5432/exampledb?connect_timeout=10&sslmode=require"
engine = create_engine(uri)
with engine.connect() as conn:
    print(conn.execute("SELECT 1").fetchone())

如果测试失败，问题出在SQLAlchemy 2.0与数据库的连接逻辑；如果成功，问题集中在Flask-Migrate的配置加载环节。

内容的提问来源于stack exchange，提问作者Nolan