import datetime from sqlalchemy import DateTime from sqlalchemy.types import TypeDecorator from lib.util_datetime import tzware_datetime from snakeeyes.extensions import db class AwareDateTime(TypeDecorator): """ A DateTime type which can only store tz-aware DateTimes. Source: https://gist.github.com/inklesspen/90b554c864b99340747e """ impl = DateTime(timezone=True) def process_bind_param(self, value, dialect): if isinstance(value, datetime.datetime) and value.tzinfo is None: raise ValueError('{!r} must be TZ-aware'.format(value)) return value def __repr__(self): return 'AwareDateTime()' class ResourceMixin(object): # Keep track when records are created and updated. created_on = db.Column(AwareDateTime(), default=tzware_datetime) updated_on = db.Column(AwareDateTime(), default=tzware_datetime, onupdate=tzware_datetime) @classmethod def sort_by(cls, field, direction): """ Validate the sort field and direction. :param field: Field name :type field: str :param direction: Direction :type direction: str :return: tuple """ if field not in cls.__table__.columns: field = 'created_on' if direction not in ('asc', 'desc'): direction = 'asc' return field, direction @classmethod def get_bulk_action_ids(cls, scope, ids, omit_ids=[], query=''): """ Determine which IDs are to be modified. :param scope: Affect all or only a subset of items :type scope: str :param ids: List of ids to be modified :type ids: list :param omit_ids: Remove 1 or more IDs from the list :type omit_ids: list :param query: Search query (if applicable) :type query: str :return: list """ # Hello. This is Nick from the future (July 2020 to be exact). This # needed to be patched to include a call to list(), otherwise it was # deleting every user instead of skipping the current_user. omit_ids = list(map(str, omit_ids)) # Hello. This is Nick from the future (July 2022 to be exact). I added # the query condition below to make this a bit more intuitive so that # an empty search query won't delete everything. Prior to this an empty # query would have been technically all of the results. if query and scope == 'all_search_results': # Change the scope to go from selected ids to all search results. ids = cls.query.with_entities(cls.id).filter(cls.search(query)) # SQLAlchemy returns back a list of tuples, we want a list of strs. ids = [str(item[0]) for item in ids] # Remove 1 or more items from the list, this could be useful in spots # where you may want to protect the current user from deleting themself # when bulk deleting user accounts. if omit_ids: ids = [id for id in ids if id not in omit_ids] return ids @classmethod def bulk_delete(cls, ids): """ Delete 1 or more model instances. :param ids: List of ids to be deleted :type ids: list :return: Number of deleted instances """ delete_count = cls.query.filter(cls.id.in_(ids)).delete( synchronize_session=False) db.session.commit() return delete_count def save(self): """ Save a model instance. :return: Model instance """ db.session.add(self) db.session.commit() return self def delete(self): """ Delete a model instance. :return: db.session.commit()'s result """ db.session.delete(self) return db.session.commit() def __str__(self): """ Create a human readable version of a class instance. :return: self """ obj_id = hex(id(self)) columns = self.__table__.c.keys() values = ', '.join("%s=%r" % (n, getattr(self, n)) for n in columns) return '<%s %s(%s)>' % (obj_id, self.__class__.__name__, values)