使用多Schema调用compare_metadata时出现AttributeError问题排查

我尝试用compare_metadata函数做元数据比较，代码结构如下：

base.py

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

workspace.py

from sqlalchemy import Column, Integer, String
from base import Base

class Workspace(Base):
    __table_args__ = {"schema": "workspace"}
    __tablename__ = 'workspace_table'
    id = Column(Integer, primary_key=True)
    name = Column(String(10))

host.py

from sqlalchemy import Column, Integer, String
from base import Base

class Host(Base):
    __table_args__ = {"schema": "host"}
    __tablename__ = 'host_table'
    id = Column(Integer, primary_key=True)
    ip = Column(String(10))

main.py

import sqlalchemy as sa
from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext
from sqlalchemy.schema import CreateSchema

import host
import workspace
from base import Base

def main():
    conn_str = "mysql+mysqlconnector://root:<my_password>@localhost"
    
    engine = sa.create_engine(conn_str)
    connection = engine.connect()
    
    connection.execute(CreateSchema("workspace", if_not_exists=True))
    connection.execute(CreateSchema("host", if_not_exists=True))
    
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
                         
    mc = MigrationContext.configure(engine.connect())
    
    diff_list = compare_metadata(mc, Base.metadata)
    
    print(diff_list)

if __name__ == "__main__":
    main()

运行main.py时触发错误：

AttributeError: 'NoneType' object has no attribute 'replace'

追踪后发现SQLAlchemy无法识别Schema，尝试给MigrationContext添加opts={"include_schemas": True}配置，问题仍未解决。

环境信息：

• macOS Sonoma
• Python 3.11
• SQLAlchemy 2.0.7
• Alembic 1.13.1
• mysql-connector-python 8.2.0

问题原因及解决方法 #

原因 #

1. 连接字符串未指定数据库：当前连接字符串仅指向MySQL服务，未指定具体操作的数据库实例，导致Schema相关操作没有明确的上下文。
2. MigrationContext配置不完整：仅添加include_schemas=True不足以让Alembic正确识别跨Schema的元数据，缺少target_metadata参数，同时未适配MySQL的Schema特性（MySQL中Schema与Database等价）。
3. Schema创建逻辑适配问题：MySQL中CREATE SCHEMA实际等同于创建数据库，而代码中模型指定的schema参数会被解析为数据库名，若未提前创建对应数据库，会导致元数据匹配失败。

解决方法 #

1. 修改连接字符串，指定目标数据库
   先确保要操作的数据库存在，再修改连接字符串：

   # 先连接到MySQL服务创建所需数据库
   root_conn_str = "mysql+mysqlconnector://root:<my_password>@localhost"
   root_engine = sa.create_engine(root_conn_str)
   with root_engine.connect() as conn:
       conn.execute(sa.text("CREATE DATABASE IF NOT EXISTS test_db"))
       conn.execute(sa.text("CREATE DATABASE IF NOT EXISTS workspace"))
       conn.execute(sa.text("CREATE DATABASE IF NOT EXISTS host"))
       conn.commit()
   
   # 连接到指定数据库
   conn_str = "mysql+mysqlconnector://root:<my_password>@localhost/test_db"
   engine = sa.create_engine(conn_str)
   

2. 完整配置MigrationContext
   创建MigrationContext时，必须传入target_metadata并开启include_schemas，确保Alembic能遍历所有指定Schema的元数据：

   with engine.connect() as connection:
       mc = MigrationContext.configure(
           connection,
           opts={
               "include_schemas": True,
               "target_metadata": Base.metadata
           }
       )
       diff_list = compare_metadata(mc, Base.metadata)
       print(diff_list)
   

3. 适配MySQL的Schema特性
   若不需要跨数据库，可将模型的schema参数移除，改用表前缀区分不同模块的表；若确实需要跨数据库，需确保MySQL用户拥有所有目标数据库的访问权限，且模型的schema参数与实际数据库名一致。

修正后的完整main.py示例 #

import sqlalchemy as sa
from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext

import host
import workspace
from base import Base

def main():
    # 初始化MySQL服务，创建所需数据库
    root_conn_str = "mysql+mysqlconnector://root:<my_password>@localhost"
    root_engine = sa.create_engine(root_conn_str)
    with root_engine.connect() as conn:
        conn.execute(sa.text("CREATE DATABASE IF NOT EXISTS test_db"))
        conn.execute(sa.text("CREATE DATABASE IF NOT EXISTS workspace"))
        conn.execute(sa.text("CREATE DATABASE IF NOT EXISTS host"))
        conn.commit()
    
    # 连接到目标数据库
    conn_str = "mysql+mysqlconnector://root:<my_password>@localhost/test_db"
    engine = sa.create_engine(conn_str)
    
    # 创建所有表
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)
                         
    # 配置MigrationContext并执行元数据比较
    with engine.connect() as connection:
        mc = MigrationContext.configure(
            connection,
            opts={
                "include_schemas": True,
                "target_metadata": Base.metadata
            }
        )
        diff_list = compare_metadata(mc, Base.metadata)
        print(diff_list)

if __name__ == "__main__":
    main()

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