initial creation

bsawf/18-processing-microtransactions/lib/util_sqlalchemy.py

@@ -0,0 +1,139 @@
+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)