Compare sqlalchemy inserts with sqlite3 executemany consuming an iterator

sa_iterparams.py

"""SQLAlalchemy inserts vs. sqlite3 lazy executemany()."""

from collections.abc import Iterator
import time

import sqlalchemy as sa
import sqlalchemy.orm

ENGINE = sa.create_engine('sqlite:///spam.sqlite3')

REGISTRY = sa.orm.registry()

N_ROWS = 1_000_000


@REGISTRY.mapped
class Spam:

    __tablename__ = 'spam'

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.Text, nullable=False)


REGISTRY.metadata.drop_all(ENGINE)

REGISTRY.metadata.create_all(ENGINE)


def insert_plain() -> None:
    query = sa.insert(Spam)
    with ENGINE.begin() as conn:
        conn.execute(query, list(iterparams_dicts()))


def iterparams_dicts(*, n: int = N_ROWS) -> Iterator[dict[str, str]]:
    for i in range(1, n + 1):
        yield {'name': f'spam-{i:d}'}


def insert_raw() -> None:
    sql = sa.insert(Spam).compile(bind=ENGINE, column_keys=['name'])
    assert sql.positional
    with ENGINE.begin() as conn:
        conn.connection.executemany(sql.string, iterparams_tuples())


def iterparams_tuples(*, n: int = N_ROWS) -> Iterator[tuple[str]]:
    for i in range(1, n + 1):
        yield (f'spam-{i:d}',)


for func in (insert_plain, insert_raw):
    start = time.perf_counter_ns()
    func()
    duration = (time.perf_counter_ns() - start) / 1_000_000_000
    print(func.__name__, f'{duration:.2f}', sep='\t')