Raw sql vs. sqlalchemy 2.0, using SQLite as DB

gistfile1.py

"""
Python 3.11.9, Win11, i5-10300H

Entries: 1000
Queries: 100

Results (in seconds):
Op         SQLAlchemy      Raw SQL         Diff      
--------------------------------------------------
Select     0.3067          0.0420          7.30      
Insert     0.0070          0.0030          2.33      
Update     0.0489          0.0000          0.00      
Delete     0.0428          0.0000          0.00      
Aggregate  0.0268          0.0070          3.83      

"""

import time
import sqlite3
from sqlalchemy import create_engine, Column, Integer, String, func
from sqlalchemy.orm import declarative_base, Session

Base = declarative_base()


class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)


def create_test_data(session, num_records):
    for i in range(num_records):
        user = User(name=f'User{i}', age=i % 100)
        session.add(user)
    session.commit()


def sqlalchemy_benchmark(session, num_queries):
    times = {
        'select': 0,
        'insert': 0,
        'update': 0,
        'delete': 0,
        'aggregate': 0
    }

    # SELECT
    start_time = time.time()
    for _ in range(num_queries):
        session.query(User).filter(User.age > 50).all()
    times['select'] = time.time() - start_time

    # INSERT
    start_time = time.time()
    for i in range(num_queries):
        user = User(name=f'NewUser{i}', age=i % 100)
        session.add(user)
    session.commit()
    times['insert'] = time.time() - start_time

    # UPDATE
    start_time = time.time()
    for i in range(num_queries):
        user = session.query(User).filter(User.id == i + 1).first()
        if user:
            user.age += 1
    session.commit()
    times['update'] = time.time() - start_time

    # DELETE
    start_time = time.time()
    for i in range(num_queries):
        user = session.query(User).filter(User.id == i + 1).first()
        if user:
            session.delete(user)
    session.commit()
    times['delete'] = time.time() - start_time

    # Aggregate Query
    start_time = time.time()
    for _ in range(num_queries):
        session.query(func.avg(User.age)).scalar()
    times['aggregate'] = time.time() - start_time

    return times


def raw_sql_benchmark(connection, num_queries):
    times = {
        'select': 0,
        'insert': 0,
        'update': 0,
        'delete': 0,
        'aggregate': 0
    }

    cursor = connection.cursor()

    # SELECT
    start_time = time.time()
    for _ in range(num_queries):
        cursor.execute("SELECT * FROM users WHERE age > 50")
        cursor.fetchall()
    times['select'] = time.time() - start_time

    # INSERT
    start_time = time.time()
    for i in range(num_queries):
        cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", (f'NewUser{i}', i % 100))
    connection.commit()
    times['insert'] = time.time() - start_time

    # UPDATE
    start_time = time.time()
    for i in range(num_queries):
        cursor.execute("UPDATE users SET age = age + 1 WHERE id = ?", (i + 1,))
    connection.commit()
    times['update'] = time.time() - start_time

    # DELETE
    start_time = time.time()
    for i in range(num_queries):
        cursor.execute("DELETE FROM users WHERE id = ?", (i + 1,))
    connection.commit()
    times['delete'] = time.time() - start_time

    # Aggregate Query
    start_time = time.time()
    for _ in range(num_queries):
        cursor.execute("SELECT AVG(age) FROM users")
        cursor.fetchone()
    times['aggregate'] = time.time() - start_time

    return times


def run_benchmark(num_records, num_queries):
    engine = create_engine('sqlite:///benchmark.db')
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        create_test_data(session, num_records)

    with Session(engine) as session:
        sqlalchemy_times = sqlalchemy_benchmark(session, num_queries)

    with sqlite3.connect('benchmark.db') as conn:
        raw_sql_times = raw_sql_benchmark(conn, num_queries)

    print(f"Entries: {num_records}")
    print(f"Queries: {num_queries}")
    print("\nResults (in seconds):")
    print("{:<10} {:<15} {:<15} {:<10}".format("Op", "SQLAlchemy", "Raw SQL", "Diff"))
    print("-" * 50)

    for operation in sqlalchemy_times.keys():
        sa_time = sqlalchemy_times[operation]
        raw_time = raw_sql_times[operation]
        diff = sa_time / raw_time if raw_time > 0 else 0
        print("{:<10} {:<15.4f} {:<15.4f} {:<10.2f}".format(
            operation.capitalize(), sa_time, raw_time, diff))


if __name__ == "__main__":
    run_benchmark(num_records=1000, num_queries=100)