A DRY ORM Implementation for Reusable CRUD Operations

[emekadefirst]

I've been using FastAPI and SQLModel to build backend APIs, and I really love the flexibility of being able to use natural Python typings, along with its default async functionality. The same goes for SQLModel — it's simple, intuitive, and powerful.

However, recently, while working with SQLModel, I found myself repeating the same pattern over and over again whenever creating a session for CRUD operations. It became quite repetitive and inefficient. The need for writing boilerplate code for each table’s CRUD operations was frustrating, and I realized there must be a better way.

So, I came up with OrmI — a DRY (Don’t Repeat Yourself) program that abstracts the session creation and CRUD logic into reusable, generic classes. This allows me to focus on the core logic of my app without having to manually set up sessions and repeat CRUD functions every time. The goal was to streamline backend development with a flexible, easy-to-use structure that works seamlessly with SQLModel and FastAPI.

👉https://github.com/emekadefirst/OrmI

[fny]

You can easily create a #query() method as part of a new base class that inherits from SQLModel:

class MyModel(SQLModel, table=True)
    @classmethod
    def query(cls, session = None): # this should be implemented in your new base class
        with Session(engine) as s:
            session = s
         return session.query(cls)

session.query is unfortunately deprecated however. I'm trying to lobby for it here: #1284 .

So instead of returning session.query(cls), I return a new AutoQuery object. Needless to say, this also incredibly redundant to implement and grow to match the wealth of statements in sqlmodel. Here it is:

# Note there are minor errors (e.g. exists)
class AutoQuery(Generic[T]):
    def __init__(self, model: type[T], session: Session):
        """Initialize with a shared session."""
        self.model = model
        self.session = session
        self.statement = select(model)

    def where(self, *conditions) -> "AutoQuery":
        """Apply WHERE conditions to the query."""
        self.statement = self.statement.where(*conditions)
        return self

    def all(self) -> list[T]:
        """Execute and return all results."""
        return self.session.exec(self.statement).all()

    def first(self) -> T | None:
        """Execute and return the first result."""
        return self.session.exec(self.statement).first()

    def last(self) -> T | None:
        """Execute and return the last result."""
        return self.session.exec(self.statement.order_by(self.model.id.desc())).first()

    def count(self) -> int:
        """Execute and return the count of results."""
        return self.session.exec(self.statement).count()

    def find(self, id: int) -> T | None:
        """Execute and return a single result by ID."""
        return self.session.exec(self.statement.where(self.model.id == id)).first()

    def exists(self) -> bool:
        """Check if any record matches the query."""
        return self.session.exec(self.statement.exists()).scalar()

    def order(self, *columns: str) -> "AutoQuery":
        """Sort results by specified columns."""
        self.statement = self.statement.order_by(*columns)
        return self

    def limit(self, n: int) -> "AutoQuery":
        """Limit the number of results returned."""
        self.statement = self.statement.limit(n)
        return self

    def offset(self, n: int) -> "AutoQuery":
        """Skip the first `n` results."""
        self.statement = self.statement.offset(n)
        return self

    def delete(self) -> int:
        """Delete matching records and return affected row count."""
        stmt = delete(self.model).where(*self.statement.whereclause)
        result = self.session.exec(stmt)
        self.session.commit()
        return result.rowcount

Now you'd simply write:

class MyModel(SQLModel, table=True)
    @classmethod
    def query(cls: type[T], session = None) -> AutoQuery[T]: # this should be implemented in your new base class
        with Session(engine) as s:
            session = s
         return AutoQuery(cls, session)

Again this is an incredible pain since there's no easy way to integrate all the queries available with select(MyModel).