Skip to content

Instantly share code, notes, and snippets.

@sirrobot01
Last active August 29, 2021 19:24
Show Gist options
  • Save sirrobot01/162fc434a8522730ab046528d5f64fdc to your computer and use it in GitHub Desktop.
Save sirrobot01/162fc434a8522730ab046528d5f64fdc to your computer and use it in GitHub Desktop.
SQLAlchemy model manager
from sqlalchemy.orm import Session
class Manager:
def __init__(self, Model, database: Session):
self.db = database
self.Model = Model
self._query = {} # Instantiate a query, update it on get/filter call
def __str__(self):
return "%s_%s" % (self.__class__.__name__, self.Model.__name__)
def __len__(self):
return self.__fetch().count()
def __iter__(self):
for obj in self.__fetch():
yield obj
def __getitem__(self, item):
return list(self)[item]
def update_query(self, query):
self._query.update(query)
def __fetch(self):
return self.db.query(self.Model).filter_by(**self._query)
def get(self, **query):
self.update_query(query)
return self.__fetch().first()
def filter(self, **query):
self.update_query(query)
return self
def create(self, **kwargs):
obj = self.Model(**kwargs)
self.save(obj)
return obj
def save(self, obj):
self.db.add(obj)
self.db.commit()
self.db.refresh(obj)
def all(self):
return self
@sirrobot01
Copy link
Author

You can use this as a manager for your SQLAlchemy model

class Market(Base):
    __tablename__ = "markets"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50))
    
    @classmethod
    def objects(cls):
        return Manager(cls, SessionLocal())

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment