Last active
July 7, 2020 15:13
-
-
Save LukeGary462/13f2c246982a2f23749f6c7dcc65a23b to your computer and use it in GitHub Desktop.
SQL Alchemy mixins I found useful for json serialization, generic queries and other boiler plate shit
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from sqlalchemy.orm.session import Session | |
| class QueryBuilder: | |
| """ | |
| This class describes a query builer. | |
| """ | |
| q_debug = False | |
| def query_from_dict(self, db_session: Session, **q_params): | |
| """ | |
| Creates a query. | |
| :param db_session: The database session | |
| :type db_session: Session | |
| :param q_params: The quarter parameters | |
| :type q_params: dictionary | |
| """ | |
| q_base = db_session.query(type(self)) | |
| for param, value in q_params.items(): | |
| if param == 'start_date': | |
| q_base = q_base.filter( | |
| type(self).__dict__.get('date_created') >= value | |
| ) | |
| elif param == 'end_date': | |
| q_base = q_base.filter( | |
| type(self).__dict__.get('date_created') <= value | |
| ) | |
| elif 'like' in param: | |
| param = param.replace('_like', '') | |
| member = type(self).__dict__.get(param) | |
| if member: | |
| q_base = q_base.filter(member.ilike(f'%{value}%')) | |
| else: | |
| q_base = q_base.filter( | |
| type(self).__dict__.get(param) == value | |
| ) | |
| if self.q_debug: | |
| print(q_base) | |
| return q_base | |
| import json | |
| from uuid import UUID | |
| from datetime import datetime | |
| from sqlalchemy.ext.declarative import DeclarativeMeta | |
| class ExtendedEncoder(json.JSONEncoder): | |
| """ | |
| This class describes an extended encoder. | |
| linter warnings disabled, this code follows the example | |
| in the official python docs, so Im keeping it the same | |
| https://mmas.github.io/sqlalchemy-serialize-json | |
| credit to Modesto Mas | |
| """ | |
| #pylint: disable=arguments-differ, method-hidden | |
| def default(self, obj): | |
| """ | |
| default method override | |
| """ | |
| if isinstance(obj, datetime): | |
| return obj.isoformat() | |
| if isinstance(obj, UUID): | |
| return str(obj) | |
| return json.JSONEncoder.default(self, obj) | |
| class JSONSerializeMixin: | |
| """ | |
| This class describes an output JSON Serialization mixin. | |
| the example from the link below is broken, this is my hacky fix | |
| https://mmas.github.io/sqlalchemy-serialize-json | |
| credit to Modesto Mas | |
| """ | |
| RELATIONSHIPS_TO_DICT = False | |
| _debug = False | |
| def __iter__(self): | |
| return self.to_dict().iteritems() | |
| @property | |
| def debug(self): | |
| ''' accessor''' | |
| return self._debug | |
| @debug.setter | |
| def debug(self, value: bool): | |
| ''' setter ''' | |
| self._debug = value | |
| def to_dict(self, **kwargs): | |
| """ | |
| serialize to dictionary for JSON usage | |
| :param rel: override class setting for relationship serialization | |
| :type rel: bool | |
| :param backref: The backref | |
| :type backref: { type_description } | |
| :returns: { description_of_the_return_value } | |
| :rtype: { return_type_description } | |
| """ | |
| _rel = kwargs.get('rel', None) | |
| _backref = kwargs.get('backref', None) | |
| _all = kwargs.get('all', False) | |
| if _rel is None: | |
| _rel = self.RELATIONSHIPS_TO_DICT | |
| res = {column.key: getattr(self, attr) | |
| for attr, column in self.__mapper__.c.items()} | |
| # look for related data | |
| if _rel: | |
| for attr, _relation in self.__mapper__.relationships.items(): | |
| if self.debug: | |
| print( | |
| f'table: {self.__table__}, ' | |
| f'attr: {attr}, ' | |
| f'rel: {_relation}, ' | |
| f'backref: {_backref}, ' | |
| ) | |
| try: | |
| # Avoid recursive loop between to tables. | |
| # hacky by looking at singular name in plural table | |
| # eg. is "core" in "cores" | |
| if f'{attr}' in f'{_backref}' and _all is False: | |
| if self.debug: | |
| print(f'skipping ref {_relation} with backref {_backref}') | |
| continue | |
| except AttributeError: | |
| pass | |
| value = getattr(self, attr) | |
| if value is None: | |
| res[_relation.key] = None | |
| elif isinstance(value.__class__, DeclarativeMeta): | |
| res[_relation.key] = value.to_dict( | |
| backref=self.__table__, | |
| **kwargs | |
| ) | |
| else: | |
| res[_relation.key] = [ | |
| i.to_dict( | |
| backref=self.__table__ | |
| ) | |
| for i in value | |
| ] | |
| return res | |
| def to_json(self, **kwargs): | |
| """ | |
| serialize to json | |
| :returns: json formatted serialized string | |
| :rtype: str | |
| """ | |
| _rel = kwargs.get('rel', None) | |
| if _rel is None: | |
| _rel = self.RELATIONSHIPS_TO_DICT | |
| return json.dumps( | |
| self.to_dict(**kwargs), | |
| cls=ExtendedEncoder | |
| ) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment