Reduce CRUD boilerplate code with a convenience mixin

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

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.

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!

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

@bolau

@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 ?

@deajan

@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='_')

@bolau

@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()

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!

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 ?

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...

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 ?

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?

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 ?

.all retrieves all objects from that class, and it's from the newer version.
My understanding is, that if you have multiple sessions at the same time and assign them to different classes in the constructor as you propose, that they won't be synced and you end up working with different cached copies of what's actually in the database. That's at least why I'm trying to use only one session in one HTTP handler.

Hmm... Indeed, I think that could be solved by using ScopedSessions, ie https://docs.sqlalchemy.org/en/14/orm/contextual.html
But then, your whole program needs to use them. So that's not elegant either.

In the end, the only solution might be to request a new session per operation, ie:

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

This is a lot of duplicated code, but at least it's for an universal boilerplate.

I've tried to find a bit of documentation about session caching.
Didn't find it to be an actual issue to re-use the same session in a class.

More interesting, as of my readings, I found that we're probably trying to reinvent the wheel here:
https://github.com/absent1706/sqlalchemy-mixins/blob/master/sqlalchemy_mixins/activerecord.py

There's already thre above full implementation of ActiveRecords for SQLAlchemy. I do think that with __config__ = None that should be SQLModel compatible.
I'll have a try next week or so

Maybe then your ActiveRecordMixin class enhancements can be added to the existing implementation ?

Interesting! I was under the impression, that it would have to be done using SQLModel classes rather than SQLAlchemy, but maybe it works on that level as well.

@bolau

@bolau Never reported back, but indeed project https://github.com/absent1706/sqlalchemy-mixins works like a charm adding ActiveRecord style syntax to SQLModel, as well as Timetamping (created_at, updated_at) and Serialization support.
I'll probably try to catch up with the author for session context integration. Thanks for the chat, and thanks for the idea you posted about SQLModel and ActiveRecords

@tiangolo Integrating https://github.com/absent1706/sqlalchemy-mixins was quite convenient in SQLModel, as long as #256 is applied. Please consider adding this directly into SQLModel as it would greatly improve the productivity ;)

@deajan

Thanks @deajan, that sounds great! Although I already have 3 projects in production with my self-made mixin, it would be nice to have an even sleeker and more complete solution. I will give it a shot! And yes, I'd also really like to see #256 being integrated as an enabler for goodies like this :)

@deajan

@deajan Before I reinvent the wheel trying to get sqlalchemy-mixins working in my application, could you share how you were able to do so? Also, did you ever make any progress on cleanly integrating the session context?

EDIT:

It looks like the answer to my first question might be as simple as:

from sqlalchemy_mixins import AllFeaturesMixin
from sqlmodel import SQLModel

# https://github.com/tiangolo/sqlmodel/issues/254
AllFeaturesMixin.__config__ = None

class BaseModel(SQLModel, AllFeaturesMixin):
    __abstract__ = True
    pass

I want to write the mixin with __config__ = None, but it cant not pass the type check.

👋 I have a wrapper library that has lifecycle hooks, ORM-like DX, easy session management, JSONB mutation tracking, and other rails-like features. Check it out! Would love to hear any feedback and have more contributions to it.

Navigation Menu

Repository navigation

Reduce CRUD boilerplate code with a convenience mixin #254

First Check

Commit to Help

Example Code

Description

Operating System

Operating System Details

SQLModel Version

Python Version

Additional Context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Reduce CRUD boilerplate code with a convenience mixin #254

Description

First Check

Commit to Help

Example Code

Description

Operating System

Operating System Details

SQLModel Version

Python Version

Additional Context

Activity

byrman commented on Feb 28, 2022

bolau commented on Feb 28, 2022

byrman commented on Feb 28, 2022

deajan commented on Sep 29, 2022

bolau commented on Sep 29, 2022

deajan commented on Sep 29, 2022

bolau commented on Sep 30, 2022

deajan commented on Sep 30, 2022

bolau commented on Sep 30, 2022

deajan commented on Sep 30, 2022

bolau commented on Sep 30, 2022

deajan commented on Sep 30, 2022

bolau commented on Sep 30, 2022

deajan commented on Sep 30, 2022

deajan commented on Sep 30, 2022

bolau commented on Sep 30, 2022

deajan commented on Nov 22, 2022

bolau commented on Nov 28, 2022

ethagnawl commented on Mar 10, 2023

baojd42 commented on Mar 31, 2023

iloveitaly commented on Apr 9, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions