+"""This series of tests illustrates different ways to INSERT a large number
+of rows in bulk.
+
+
+"""
from . import Profiler
from sqlalchemy.ext.declarative import declarative_base
@Profiler.profile
def test_bulk_save(n):
- """Batched INSERT statements via the ORM in "bulk", discarding PK values."""
+ """Batched INSERT statements via the ORM in "bulk", discarding PKs."""
session = Session(bind=engine)
session.bulk_save_objects([
Customer(
@Profiler.profile
def test_bulk_insert_mappings(n):
- """Batched INSERT statements via the ORM "bulk", using dictionaries instead of objects"""
+ """Batched INSERT statements via the ORM "bulk", using dictionaries."""
session = Session(bind=engine)
session.bulk_insert_mappings(Customer, [
dict(
@Profiler.profile
def test_dbapi_raw(n):
- """The DBAPI's pure C API inserting rows in bulk, no pure Python at all"""
+ """The DBAPI's API inserting rows in bulk."""
conn = engine.pool._creator()
cursor = conn.cursor()
@Profiler.profile
def test_dbapi_fetchall_plus_append_objects(n):
- """Load rows using DBAPI fetchall(), make a list of objects."""
+ """Load rows using DBAPI fetchall(), generate an object for each row."""
_test_dbapi_raw(n, True)
cursor.execute(sql)
if make_objects:
- result = []
for row in cursor.fetchall():
# ensure that we fully fetch!
customer = SimpleCustomer(
id=row[0], name=row[1], description=row[2])
- result.append(customer)
else:
for row in cursor.fetchall():
# ensure that we fully fetch!