SQLAlchemy ORM 使用工作单元同步数据库更改时的模式。这种模式远远超出了简单的数据“插入”范围。它包括使用属性检测系统接收分配给对象的属性，该系统跟踪对象的更改，包括在对象中跟踪插入的所有行。恒等映射其效果是，对于每一行，SQLAlchemy 必须检索其“最后插入的 id”（如果尚未给出），并且还涉及根据需要扫描要插入的行并对其依赖关系进行排序。为了保持所有这些运行，对象还需要进行相当程度的簿记，对于大量的行来说，这可能会在大型数据结构上花费过多的时间，因此最好将它们分块。
基本上，工作单元是很大程度的自动化，以便自动执行将复杂对象图持久化到关系数据库中的任务，而无需显式持久化代码，并且这种自动化是有代价的。
因此 ORM 基本上不适合高性能批量插入。这就是 SQLAlchemy 的全部原因two单独的库，如果您查看的话您会注意到http://docs.sqlalchemy.org/en/latest/index.html您将在索引页中看到两个不同的部分 - 一个用于 ORM，另一个用于 Core。如果不了解两者，就无法有效地使用 SQLAlchemy。
对于快速批量插入的用例，SQLAlchemy 提供了core，这是 ORM 构建于其之上的 SQL 生成和执行系统。有效地使用这个系统，我们可以生成与原始 SQLite 版本竞争的 INSERT。下面的脚本说明了这一点，以及预先分配主键标识符的 ORM 版本，以便 ORM 可以使用executemany() 来插入行。两个 ORM 版本都一次对 1000 条记录进行分块刷新，这对性能有显着影响。
这里观察到的运行时间是：
SqlAlchemy ORM: Total time for 100000 records 16.4133379459 secs
SqlAlchemy ORM pk given: Total time for 100000 records 9.77570986748 secs
SqlAlchemy Core: Total time for 100000 records 0.568737983704 secs
sqlite3: Total time for 100000 records 0.595796823502 sec
script:
import time
import sqlite3
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
Base = declarative_base()
DBSession = scoped_session(sessionmaker())
class Customer(Base):
__tablename__ = "customer"
id = Column(Integer, primary_key=True)
name = Column(String(255))
def init_sqlalchemy(dbname = 'sqlite:///sqlalchemy.db'):
global engine
engine = create_engine(dbname, echo=False)
DBSession.remove()
DBSession.configure(bind=engine, autoflush=False, expire_on_commit=False)
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
def test_sqlalchemy_orm(n=100000):
init_sqlalchemy()
t0 = time.time()
for i in range(n):
customer = Customer()
customer.name = 'NAME ' + str(i)
DBSession.add(customer)
if i % 1000 == 0:
DBSession.flush()
DBSession.commit()
print "SqlAlchemy ORM: Total time for " + str(n) + " records " + str(time.time() - t0) + " secs"
def test_sqlalchemy_orm_pk_given(n=100000):
init_sqlalchemy()
t0 = time.time()
for i in range(n):
customer = Customer(id=i+1, name="NAME " + str(i))
DBSession.add(customer)
if i % 1000 == 0:
DBSession.flush()
DBSession.commit()
print "SqlAlchemy ORM pk given: Total time for " + str(n) + " records " + str(time.time() - t0) + " secs"
def test_sqlalchemy_core(n=100000):
init_sqlalchemy()
t0 = time.time()
engine.execute(
Customer.__table__.insert(),
[{"name":'NAME ' + str(i)} for i in range(n)]
)
print "SqlAlchemy Core: Total time for " + str(n) + " records " + str(time.time() - t0) + " secs"
def init_sqlite3(dbname):
conn = sqlite3.connect(dbname)
c = conn.cursor()
c.execute("DROP TABLE IF EXISTS customer")
c.execute("CREATE TABLE customer (id INTEGER NOT NULL, name VARCHAR(255), PRIMARY KEY(id))")
conn.commit()
return conn
def test_sqlite3(n=100000, dbname = 'sqlite3.db'):
conn = init_sqlite3(dbname)
c = conn.cursor()
t0 = time.time()
for i in range(n):
row = ('NAME ' + str(i),)
c.execute("INSERT INTO customer (name) VALUES (?)", row)
conn.commit()
print "sqlite3: Total time for " + str(n) + " records " + str(time.time() - t0) + " sec"
if __name__ == '__main__':
test_sqlalchemy_orm(100000)
test_sqlalchemy_orm_pk_given(100000)
test_sqlalchemy_core(100000)
test_sqlite3(100000)
也可以看看：http://docs.sqlalchemy.org/en/latest/faq/performance.html
