-
-
Notifications
You must be signed in to change notification settings - Fork 623
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Reduce CRUD boilerplate code with a convenience mixin #254
Comments
Thanks for sharing. However, I would not consider your
This failed, however, with: |
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. |
@deajan Great, thanks for letting me know! 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 Thanks, indeed it looks like alot of work went into that boilerplate. I still have a quick question. Having to provide a session for each function call is alot of work. Why not make session optional ? @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 Also, we might just use @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. What are your thoughts on this ? |
I swapped session to be the first parameter, since I passed it everytime - makes the method signatures look
So in my ideal world, the ActiveRecord methods should create a new session with |
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. 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 Maybe there's something I am missing ? |
|
Hmm... Indeed, I think that could be solved by using ScopedSessions, ie https://docs.sqlalchemy.org/en/14/orm/contextual.html 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. More interesting, as of my readings, I found that we're probably trying to reinvent the wheel here: There's already thre above full implementation of ActiveRecords for SQLAlchemy. I do think that with 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 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. @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 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:
|
First Check
Commit to Help
Example Code
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
The text was updated successfully, but these errors were encountered: