python sqlite orm benchmark

python_orm_bench.py

"""
py3.8 - macbook pro 16-inch, 2019, 2.3 GHz 8-Core Intel Core i9, 16 GB 2667 MHz DDR4

In memory benchmarks, working with 100_000 items at a time

SqlAlchemy - insert bulk: 1560.605ms
SqlAlchemy - select all: 1350.774ms
SqlAlchemy - insert singles: 7821.888ms

sqlite3 - insert singles: 140.412ms
sqlite3 - select all: 62.951ms
sqlite3 - insert bulk: 108.903ms

peewee - insert bulk: 2492.869ms
peewee - select all: 774.247ms
peewee - insert batch (4096): 2295.691ms
peewee - insert batch (1024): 1961.476ms
peewee - insert batch (256): 1706.937ms
peewee - insert batch (64): 1837.337ms
peewee - insert batch (16): 2433.151ms
peewee - insert batch (4): 4391.415ms
peewee - insert singles: 10709.015ms

pony - insert bulk: 2918.317ms
pony - select all: 1867.465ms
pony - insert singles: 8605.021ms
"""

import contextlib
import time
import sqlite3


@contextlib.contextmanager
def Timer(name: str):
    t = time.time_ns()
    try:
        yield
    finally:
        t = (time.time_ns() - t) / 1000_000
        print(f'{name}: {t:.3f}ms')


def run_test_sqlalchemy(n=100000, dbname='sqlite:///:memory:'):
    import sqlalchemy as sa
    import sqlalchemy.orm as sao

    # init
    base = sao.declarative_base()
    db_session = sao.scoped_session(sao.sessionmaker())

    class Customer(base):
        __tablename__ = "customer"
        id = sa.Column(sa.Integer, primary_key=True)
        name = sa.Column(sa.String(255))

    engine = sa.create_engine(dbname, echo=False)
    db_session.configure(bind=engine, autoflush=False, expire_on_commit=False)
    base.metadata.drop_all(engine)
    base.metadata.create_all(engine)

    # tests
    with Timer('SqlAlchemy - insert bulk'):
        db_session.bulk_save_objects(Customer(name='NAME ' + str(i)) for i in range(n))
        db_session.commit()

    with Timer('SqlAlchemy - select all'):
        for i, item in enumerate(db_session.query(Customer).all()):
            pass
    print(f'- count: {i + 1}')

    with Timer('SqlAlchemy - insert singles'):
        for i in range(n):
            db_session.add(Customer(name='NAME ' + str(i)))
        db_session.commit()




def run_test_sqlite3(n=100000, dbname=':memory:'):
    # init
    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()

    # tests
    c = conn.cursor()

    with Timer('sqlite3 - insert singles'):
        for i in range(n):
            row = ('NAME ' + str(i),)
            c.execute("INSERT INTO customer (name) VALUES (?)", row)
        conn.commit()

    with Timer('sqlite3 - select all'):
        for i, item in enumerate(c.execute("SELECT * FROM customer")):
            pass
    print(f'- count: {i + 1}')

    with Timer('sqlite3 - insert bulk'):
        c.executemany("INSERT INTO customer (name) VALUES (?)", (('NAME ' + str(i),) for i in range(n)))
        conn.commit()



def run_test_peewee(n=100000, dbname=':memory:'):
    import peewee

    # init
    db = peewee.SqliteDatabase(dbname, pragmas={'foreign_keys': 1})

    class Customer(peewee.Model):
        id = peewee.IntegerField(primary_key=True)
        name = peewee.CharField(max_length=255)
        class Meta:
            database = db

    Customer.create_table()

    # tests
    with Timer('peewee - insert bulk'):
        Customer.bulk_create([Customer(name='NAME ' + str(i)) for i in range(n)])

    with Timer('peewee - select all'):
        for i, item in enumerate(Customer.select()):
            pass
    print(f'- count: {i + 1}')

    for b in [4096, 1024, 256, 64, 16, 4]:
        with Timer(f'peewee - insert batch ({b})'):
            Customer.bulk_create([Customer(name='NAME ' + str(i)) for i in range(n)], batch_size=b)

    with Timer('peewee - insert singles'):
        for i in range(n):
            Customer.create(name='NAME ' + str(i))



def run_test_pony(n=100000, dbname=':memory:'):
    import pony.orm as p

    # init
    db = p.Database()

    class Customer(db.Entity):
        id = p.PrimaryKey(int, auto=True)
        name = p.Required(str)

    db.bind(provider='sqlite', filename=dbname, create_db=True)
    db.generate_mapping(create_tables=True)

    # tests
    with Timer('pony - insert bulk'):
        with p.db_session:
            for i in range(n):
                customer = Customer(name='NAME ' + str(i))

    with Timer('pony - select all'):
        with p.db_session:
            for i, item in enumerate(p.select(c for c in Customer)):
                pass
    print(f'- count: {i+1}')

    with Timer('pony - insert singles'):
        for i in range(n):
            with p.db_session:
                customer = Customer(name='NAME ' + str(i))


if __name__ == '__main__':
    run_test_sqlalchemy(100_000, dbname='sqlite:///:memory:')
    print()
    run_test_sqlite3(100_000, dbname=':memory:')
    print()
    run_test_peewee(100_000, dbname=':memory:')
    print()
    run_test_pony(100_000, dbname=':memory:')
    print()