Python

Python clean code tip - always use paginated queries

Python clean code tip:

Always use paginated queriesl

Use the "last evaluated record" approach to paginate instead of offset. This way you limit database load per single query.

👇

from sqlalchemy.orm import Session, sessionmaker, declarative_base
from sqlalchemy import create_engine, Column, Integer, String


SQLALCHEMY_DATABASE_URL = "sqlite:///./example.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)

    def __repr__(self):
        return f"User(id={self.id}, email={self.email})"


class SQLiteUserRepository:
    BATCH_SIZE = 1000

    def __init__(self, database_session):
        self._database_session = database_session

    def add(self, user):
        with self._database_session.begin():
            self._database_session.add(user)

    def list_all(self):
        last_fetched_id = -1
        users = []

        while True:
            users_batch = (
                self._database_session.query(User)
                .filter(User.id > last_fetched_id)
                .order_by(User.id)
                .limit(self.BATCH_SIZE)
                .all()
            )
            users.extend(users_batch)
            if not users_batch:
                break
            last_fetched_id = users_batch[-1].id

        return users


Base.metadata.create_all(bind=engine)
database_session = sessionmaker(engine)()

repository = SQLiteUserRepository(database_session)
repository.add(User(email="[email protected]"))
repository.add(User(email="[email protected]"))

print(repository.list_all())
# [User(id=1, [email protected]), User(id=2, [email protected])]