Reduce CRUD boilerplate code with a convenience mixin

[bolau]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from typing import Optional, Union
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Field, Session, SQLModel, create_engine, select


class ActiveRecord(SQLModel):
    @classmethod
    def by_id(cls, id: int, session):
        obj = session.get(cls, id)
        if obj is None:
            raise HTTPException(status_code=404, detail=f"{cls.__name__} with id {id} not found")
        return obj

    @classmethod
    def all(cls, session):
        return session.exec(select(cls)).all()

    @classmethod
    def create(cls, source: Union[dict, SQLModel], session):
        if isinstance(source, SQLModel):
            obj = cls.from_orm(source)
        elif isinstance(source, dict):
            obj = cls.parse_obj(source)
        session.add(obj)
        session.commit()
        session.refresh(obj)
        return obj

    def save(self, session):
        session.add(self)
        session.commit()
        session.refresh(self)

    def update(self, source: Union[dict, SQLModel], session):
        if isinstance(source, SQLModel):
            source = source.dict(exclude_unset=True)

        for key, value in source.items():
            setattr(self, key, value)
        self.save(session)

    def delete(self, session):
        session.delete(self)
        session.commit()


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

class HeroCreate(HeroBase):
    pass

class HeroRead(HeroBase):
    id: int

class HeroUpdate(SQLModel):
    name: Optional[str] = None
    secret_name: Optional[str] = None
    age: Optional[int] = None

class Hero(HeroBase, ActiveRecord, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session


app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()

@app.post("/heroes/", response_model=HeroRead)
def create_hero(hero: HeroCreate, session: Session = Depends(get_session)):
    return Hero.create(hero, session)

@app.get("/heroes/", response_model=list[HeroRead])
def read_heroes(session: Session = Depends(get_session)):
    return Hero.all(session)

@app.get("/heroes/{id}", response_model=HeroRead)
def read_heroes(id: int, session: Session = Depends(get_session)):
    return Hero.by_id(id, session)

@app.patch("/heroes/{id}", response_model=HeroRead)
def read_heroes(id: int, hero: HeroUpdate, session: Session = Depends(get_session)):
    db_hero = Hero.by_id(id, session)
    db_hero.update(hero, session)
    return db_hero

Description

Hi all,

I'm fairly new to FastAPI and SQLModel, but I really like these libraries. After porting two Flask+SQLAlchemy projects to FastAPI+SQLModel, I noticed that I write lots of boilerplate code to implement basic model functions like create, save, delete or update. I was surprised that the examples in the documentation implement CRUD functionality with functions rather than member functions of the models.

So I wrote a class called ActiveRecord to be used as a mixin for SQLModel classes, which adds these functions and can be reused for different models and projects. In my code above you see it being used with the multi-model Hero example in the documentation, which shortens the remaining code (compare https://sqlmodel.tiangolo.com/tutorial/fastapi/multiple-models)

Using something like this in all of my SQLModel classes where table=True seems so obvious, that I suspect that there's either something I'm missing here, or this is functionality that's really missing in SQLModel. Why doesn't the SQLModel class have such functions? How do you do that? Do you think this is worth contributing? I guess, at least I could clean it up and share it as gist.

Thanks for any feedback,
Boris

Operating System

macOS

Operating System Details

No response

SQLModel Version

0.0.6

Python Version

3.9.5

Additional Context

No response

[byrman]

Thanks for sharing. However, I would not consider your ActiveRecord to be a (true) mixin: it's just a subclass of SQLModel. So, I tried to improve upon your code like this:

class ActiveRecord:
    pass


class Hero(ActiveRecord, HeroBase, table=True):
    pass

This failed, however, with: AttributeError: type object 'ActiveRecord' has no attribute '__config__'. This might be a bug in SQLModel and I will make a pull request for it.

[bolau]

You're totally right. I ran into the same error, that's why I subclassed it, although it shouldn't be necessary in theory. But I wasn't able to solve this in a cleaner way. If you can fix that, that would be awesome!

[byrman]

This pull request would add support for mixins: #256.

[deajan]

@bolau I really enjoy your ActiveRecord class.
I really think this should be part of sqlmodel basics, since it really scaffolds REST usage nicely.
@tiangolo Would you consider adding this to the SQLModel base class ?

[bolau]

@deajan Great, thanks for letting me know!
Here's an updated version that I currently use:

import logging
from typing import Union, Optional, Any

import pandas as pd
from sqlmodel import SQLModel, select
from sqlalchemy.exc import IntegrityError, NoResultFound, OperationalError
from sqlalchemy.orm.exc import FlushError


class ActiveRecordMixin():
    __config__ = None

    @property
    def primary_key(self):
        return self.__mapper__.primary_key_from_instance(self)

    @classmethod
    def first(cls, session):
        statement = select(cls)
        return session.exec(statement).first()

    @classmethod
    def one_by_id(cls, session, id: int):
        obj = session.get(cls, id)
        return obj

    @classmethod
    def first_by_field(cls, session, field: str, value: Any):
        return cls.first_by_fields(session, {field: value})

    @classmethod
    def one_by_field(cls, session, field: str, value: Any):
        return cls.one_by_fields(session, {field: value})

    @classmethod
    def first_by_fields(cls, session, fields: dict):
        statement = select(cls)
        for key, value in fields.items():
            statement = statement.where(getattr(cls, key) == value)
        try:
            return session.exec(statement).first()
        except NoResultFound:
            logging.error(f"{cls}: first_by_fields failed, NoResultFound")
            return None

    @classmethod
    def one_by_fields(cls, session, fields: dict):
        statement = select(cls)
        for key, value in fields.items():
            statement = statement.where(getattr(cls, key) == value)
        try:
            return session.exec(statement).one()
        except NoResultFound:
            logging.error(f"{cls}: one_by_fields failed, NoResultFound")
            return None

    @classmethod
    def all_by_field(cls, session, field: str, value: Any):
        statement = select(cls).where(getattr(cls, field) == value)
        return session.exec(statement).all()

    @classmethod
    def all_by_fields(cls, session, fields: dict):
        statement = select(cls)
        for key, value in fields.items():
            statement = statement.where(getattr(cls, key) == value)
        return session.exec(statement).all()

    @classmethod
    def convert_without_saving(cls, source: Union[dict, SQLModel], update: Optional[dict] = None) -> SQLModel:
        # try:
        if isinstance(source, SQLModel):
            obj = cls.from_orm(source, update=update)
        elif isinstance(source, dict):
            obj = cls.parse_obj(source, update=update)
        # except ValidationError:
        #    return None
        return obj

    @classmethod
    def create(cls, session, source: Union[dict, SQLModel], update: Optional[dict] = None) -> Optional[SQLModel]:
        obj = cls.convert_without_saving(source, update)
        if obj is None:
            return None
        if obj.save(session):
            return obj
        return None

    @classmethod
    def create_or_update(cls, session, source: Union[dict, SQLModel], update: Optional[dict] = None)\
            -> Optional[SQLModel]:
        obj = cls.convert_without_saving(source, update)
        if obj is None:
            return None
        pk = cls.__mapper__.primary_key_from_instance(obj)
        if pk[0] is not None:
            existing = session.get(cls, pk)
            if existing is None:
                return None  # Error
            else:
                existing.update(session, obj)  # Update
                return existing
        else:
            return cls.create(session, obj)  # Create

    @classmethod
    def count(cls, session) -> int:
        return len(cls.all(session))

    def refresh(self, session):
        session.refresh(self)

    def save(self, session) -> bool:
        session.add(self)
        try:
            session.commit()
            session.refresh(self)
            return True
        except (IntegrityError, OperationalError, FlushError) as e:
            logging.error(e)
            session.rollback()
            return False

    def update(self, session, source: Union[dict, SQLModel]):
        if isinstance(source, SQLModel):
            source = source.dict(exclude_unset=True)

        for key, value in source.items():
            setattr(self, key, value)
        self.save(session)

    def delete(self, session):
        session.delete(self)
        session.commit()

    @classmethod
    def all(cls, session):
        return session.exec(select(cls)).all()

    @classmethod
    def delete_all(cls, session):
        for obj in cls.all(session):
            obj.delete(session)

    @classmethod
    def to_pandas(cls, session) -> pd.DataFrame:
        records = cls.all(session)
        return pd.json_normalize([r.dict() for r in records], sep='_')

[deajan]

@bolau Thanks, indeed it looks like alot of work went into that boilerplate.
Apart from the pandas function which comment out, it's quite enjoyable.

I still have a quick question. Having to provide a session for each function call is alot of work.
If you're using fastapi's depends() functionality, it might make sense.
But without it, for every member function call we need a with get_session() as session context statement before using the ActiveRecord functions.

Why not make session optional ?
Eg your current code:

    @classmethod
    def one_by_id(cls, session, id: int):
        obj = session.get(cls, id)
        return obj

would become

    @classmethod
    def one_by_id(cls, id: int, session = None):
        if not session:
            session = get_session()
        obj = session.get(cls, id)
        return obj

With get_session() being the boring standard, eg:

from sqlmodel import SQLModel, Session, create_engine
from contextlib import contextmanager
from logging import getLogger

@contextmanager
def get_session():
    session = Session(engine)
    try:
        yield session
        session.commit()
    except Exception as exc:
        logger.error("SQL Error: %s" % exc.__str__())
        logger.debug("Trace:", exc_info=True)
        session.rollback()
        logger.error("SQL Error: Rollback complete.")
        raise
    finally:
        session.close()

[bolau]

That's a fabulous idea! Having to pass the session annoyed me, too. I was under the impression, that I should really try to use only a single session per HTTP handler for all ActiveRecord actions. But as a FastAPI and SQLModel newbie, I didn't know how to solve that. Your approach is very pragmatic, since it just acquires a session in every call. And maybe that's the right thing to do!

[deajan]

This would indeed make all function signatures swap parameters since session must be the last one since it's optional.
I noticed that you already swapped function signatures between ActiveRecord and ActiveRecordMixin classes.
Eg for by_id functions, the signature was (int, session) -> results and became (session, int).
Was there any special reason for that ?

Also, we might just use **kwargs as last argument maybe, something that would look alike:

@classmethod
    def one_by_id(cls, id: int, **kwargs):
        session = kwargs.pop("session", get_session())
        obj = session.get(cls, id)
        return obj

The drawback would be that you could pass virtually any parameter to the class methods without any noticeable errors.
There could also be a solution for this problem, by raising an error when kwargs > 0 once session has been taken from it.

What are your thoughts on this ?

[bolau]

I swapped session to be the first parameter, since I passed it everytime - makes the method signatures look
more similar. I'd prefer your session=None over the kwargs. But IMHO, the nicest solution from a user's standpoint would be if the correct session is "just there" as a global object, without passing it explicitly. Similar to flask.request:

flask.request
To access incoming request data, you can use the global request object. Flask parses incoming request data for you and gives you access to it through that global object. Internally Flask makes sure that you always get the correct data for the active thread if you are in a multithreaded environment.

So in my ideal world, the ActiveRecord methods should create a new session with get_session if there is none open, and use the existing session when being called inside a session context, but without explicitly receiving the session as an argument.
But I don't know how to do that. I've tried it with FastAPI.Depends, but I realized that those don't work that way...

[deajan]

The problem with the global object is that you have class methods as well as instance methods mixed in ActiveRecord class, so there would be at least two "global" sessions, one for the instance and one for the class.

So far, here's what I did:

class ActiveRecordMixin():
    __config__ = None

    @classmethod
    def __init__(cls, *args, **kwargs):
        super(ActiveRecordMixin).__init__(*args, **kwargs)
        cls.session = kwargs.pop("session", get_session())

    @classmethod
    def one_by_id(cls, id: int):
        with cls.session as session:
            obj = session.get(cls, id)
        return obj

I am thinking of a way to not duplicate the classmethod code to the instance. Any ideas perhaps ?

[bolau]

Hm, does this really work? When you use the class in two different handlers, it will use the same session. I thought the session has to be more 'local' to the handler, like this:

@router.get("/myobject")
def my_route(session: Session = Depends(get_session)):
    return MyObject.all()    <--- this should use the session obtained here

Doesn't that matter?

[deajan]

Not sure to understand what you mean.
Is the all() function from earlier ActiveRecord class you posted ?

So yes, I'm using the same session for multiple instances of the class.

For the example I gave above, I don't care about getting the session via FastAPI's Depends(). I just use the session obtained in the object itself. AFAIK there's no reason I'd need to use the FastAPI obtained session instead of the object's one.

Maybe there's something I am missing ?