python - SQLAlchemy bulk_insert_mappings() : Could not get mapper for table 'test'

Question

我一直在尝试使用 sqlalchemy 的 bulk_insert_mappings。我知道我可以创建 session 并连接到数据库。我已经初始化了我的引擎，但我似乎无法从表中获取我需要的映射器。

from sqlalchemy import create_engine
from sqlalchemy.orm.session import sessionmaker,Session
from sqlalchemy_utils import get_mapper

engine = create_engine('mysql+pymysql://{}:{}@IP:PORT/'.format(USER,PW)) # removed my config here
connection = engine.connect()
m = MetaData(bind=engine,schema='test')
m.reflect()

Session = sessionmaker(bind=engine)
s = Session()
s.bulk_insert_mappings(get_mapper(m.tables['test.test']), pd.DataFrame({'a':['a','b','c']}).to_dict(orient="records"))
s.commit()
s.close()

我最近在SO上发现了一堆相关的问题

SQLAlchemy get Mapper object from Table object (from Metadata or Session or otherwise)

但是 sqlalchemy_utils.get_mapper 提出:

"ValueError: Could not get mapper for table 'test'."

sqlalchemy.orm.mapperlib._mapper_registry 似乎是空的。也许是因为我没有将它绑定(bind)到我的引擎。但不确定该怎么做。

PS:test是一个很简单的TEXT类型的一列表

这是 m.tables['test.test'] 的输出

Table('test', MetaData(bind=Engine(mysql+pymysql://USER:***@IP:PORT/)), Column('a', TEXT(), table=<test>), schema='test')

Answer 1

SQLAlchemy 的工作 Mapper是为了:

Define the correlation of class attributes to database table columns.

... 它是 SQLAlchemy ORM 的基础。使用 ORM，Python 类表示数据库中的表，并且需要某种机制将类的属性与表中的列相关联。如果您不使用 ORM，则您的表不会映射到 Python 类，因此没有使用映射器。这就是您从 get_mapper() 中得到错误的原因。

在你的例子中:

m = MetaData(bind=engine,schema='test')
m.reflect()

MetaData是:

A collection of Table objects and their associated schema constructs.

和MetaData.reflect :

Automatically creates Table entries in this MetaData for any table available in the database but not yet present in the MetaData.

此时，您有一组 Table 对象，您想要对其中一个执行批量插入。不要将 Table 对象与 ORM 映射类混淆，它们不是一回事。

bulk_insert_mappings 上的文档状态:

Perform a bulk insert of the given list of mapping dictionaries.

和

The values within the dictionaries as given are typically passed without modification into Core Insert() constructs

您正在尝试实现数据的批量插入，我们可以跳过 ORM 方法(任何涉及 Session 的方法)并显式地与 Core 交互。

表达式 pd.DataFrame({'a':['a','b','c']}).to_dict(orient="records") 返回 dict 就像:[{'a': 'a'}, {'a': 'b'}, {'a': 'c'}], 所以为了简单起见，我将使用此处的示例输出。

您的元数据对象中有您已使用 m.tables['test.test'] 检索的表，并且该 Table 对象可用于生成它自己的插入语句:

print(m.tables['test.test'].insert())
# INSERT INTO test.test (a) VALUES (%(a)s)

为了执行多个语句，我们可以将字典列表传递给 Connection.execute()，如下所示。

ORM Session 的好处之一是它允许显式事务管理，您可以在其中调用 Session.rollback() 或 Session.commit( ) 必要时。连接对象也可以使用 Engine.begin() 在类似于 Session 的事务中显式操作.

例如，使用上下文管理器:

with engine.begin() as conn:
    conn.execute(
        m.tables['test.test'].insert(),
        *[{'a': 'a'}, {'a': 'b'}, {'a': 'c'}]
    )

如果上下文中没有错误，这将自动提交查询，如果有错误则回滚。

引擎日志显示此表达式发出以下查询:

INSERT INTO test.test (a) VALUES (%(a)s)
({'a': 'a'}, {'a': 'b'}, {'a': 'c'})

以下人为设计的示例通过使用 Session.bulk_insert_mappings() 显示您的原始查询。我不得不创建一个 ORM 模型来表示该表并向该表添加一个 id 字段，因为 ORM 不喜欢在没有主键的情况下工作。

m = MetaData(bind=engine,schema='test')
Base = declarative_base(metadata=m)

class Test(Base):
    __tablename__ = 'test'
    id = Column(Integer, primary_key=True)
    a = Column(Text)


Session = sessionmaker(bind=engine)
s = Session()
s.bulk_insert_mappings(get_mapper(m.tables['test.test']), pd.DataFrame({'a':['a','b','c']}).to_dict(orient="records"))
s.commit()
s.close()

这是从引擎日志中执行的查询:

INSERT INTO test.test (a) VALUES (%(a)s)
({'a': 'a'}, {'a': 'b'}, {'a': 'c'})

您会注意到，这与我们通过直接使用 Core 能够实现的查询完全相同。