[SDESK-7386] Support async app.on_ events

eve/events.py

@@ -0,0 +1,128 @@
+# -*- coding: utf-8 -*-
+
+"""
+    Events
+    ~~~~~~
+
+    Implements C#-Style Events.
+
+    Derived from the original work by Zoran Isailovski:
+    http://code.activestate.com/recipes/410686/ - Copyright (c) 2005
+
+    Code copied from original Events library into this library, to support async
+    https://github.com/pyeve/events
+
+    :copyright: (c) 2014-2017 by Nicola Iarocci.
+    :license: BSD, see LICENSE for more details.
+"""
+
+from inspect import iscoroutinefunction
+
+
+class EventsException(Exception):
+    pass
+
+
+class Events:
+    """
+    Encapsulates the core to event subscription and event firing, and feels
+    like a "natural" part of the language.
+
+    The class Events is there mainly for 3 reasons:
+
+        - Events (Slots) are added automatically, so there is no need to
+        declare/create them separately. This is great for prototyping. (Note
+        that `__events__` is optional and should primarilly help detect
+        misspelled event names.)
+        - To provide (and encapsulate) some level of introspection.
+        - To "steel the name" and hereby remove unneeded redundancy in a call
+        like:
+
+            xxx.OnChange = event('OnChange')
+    """
+    def __init__(self, events=None):
+
+        if events is not None:
+
+            try:
+                for _ in events:
+                    break
+            except:
+                raise AttributeError("type object %s is not iterable" %
+                                     (type(events)))
+            else:
+                self.__events__ = events
+
+    def __getattr__(self, name):
+        if name.startswith('__'):
+            raise AttributeError("type object '%s' has no attribute '%s'" %
+                                 (self.__class__.__name__, name))
+
+        if hasattr(self, '__events__'):
+            if name not in self.__events__:
+                raise EventsException("Event '%s' is not declared" % name)
+
+        elif hasattr(self.__class__, '__events__'):
+            if name not in self.__class__.__events__:
+                raise EventsException("Event '%s' is not declared" % name)
+
+        self.__dict__[name] = ev = _EventSlot(name)
+        return ev
+
+    def __repr__(self):
+        return '<%s.%s object at %s>' % (self.__class__.__module__,
+                                         self.__class__.__name__,
+                                         hex(id(self)))
+
+    __str__ = __repr__
+
+    def __len__(self):
+        return len(self.__dict__.items())
+
+    def __iter__(self):
+        def gen(dictitems=self.__dict__.items()):
+            for attr, val in dictitems:
+                if isinstance(val, _EventSlot):
+                    yield val
+        return gen()
+
+
+class _EventSlot:
+    def __init__(self, name):
+        self.targets = []
+        self.__name__ = name
+
+    def __repr__(self):
+        return "event '%s'" % self.__name__
+
+    def __call__(self, *a, **kw):
+        for f in tuple(self.targets):
+            f(*a, **kw)
+
+    async def call_async(self, *a, **kw):
+        for f in tuple(self.targets):
+            if iscoroutinefunction(f):
+                await f(*a, **kw)
+            else:
+                f(*a, **kw)
+
+    def __iadd__(self, f):
+        self.targets.append(f)
+        return self
+
+    def __isub__(self, f):
+        while f in self.targets:
+            self.targets.remove(f)
+        return self
+
+    def __len__(self):
+        return len(self.targets)
+
+    def __iter__(self):
+        def gen():
+            for target in self.targets:
+                yield target
+        return gen()
+
+    def __getitem__(self, key):
+        return self.targets[key]

eve/flaskapp.py

@@ -15,7 +15,7 @@
 import sys
 import warnings
 
-from events import Events
+from .events import Events
 from quart import Quart
 from werkzeug.routing import BaseConverter
 

eve/methods/common.py

@@ -1335,7 +1335,7 @@ def pre_event(f):
     """
 
     @wraps(f)
-    def decorated(*args, **kwargs):
+    async def decorated(*args, **kwargs):
         method = request.method
         if method == "HEAD":
             method = "GET"
@@ -1358,12 +1358,12 @@ def decorated(*args, **kwargs):
             rh_params = (request,)
 
         # general hook
-        getattr(app, event_name)(*gh_params)
+        await getattr(app, event_name).call_async(*gh_params)
         if resource:
             # resource hook
-            getattr(app, event_name + "_" + resource)(*rh_params)
+            await getattr(app, event_name + "_" + resource).call_async(*rh_params)
 
-        r = f(resource, **combined_args)
+        r = await f(resource, **combined_args)
         return r
 
     return decorated
@@ -1436,7 +1436,7 @@ def strip_prefix(hit):
     return path
 
 
-def oplog_push(resource, document, op, id=None):
+async def oplog_push(resource, document, op, id=None):
     """Pushes an edit operation to the oplog if included in OPLOG_METHODS. To
     save on storage space (at least on MongoDB) field names are shortened:
 
@@ -1523,7 +1523,7 @@ def oplog_push(resource, document, op, id=None):
 
     if entries:
         # notify callbacks
-        getattr(app, "on_oplog_push")(resource, entries)
+        await getattr(app, "on_oplog_push").call_async(resource, entries)
         # oplog push
         app.data.insert(config.OPLOG_NAME, entries)
 

eve/methods/delete.py

@@ -102,8 +102,8 @@ async def deleteitem_internal(
 
     # notify callbacks
     if not suppress_callbacks:
-        getattr(app, "on_delete_item")(resource, original)
-        getattr(app, "on_delete_item_%s" % resource)(original)
+        await getattr(app, "on_delete_item").call_async(resource, original)
+        await getattr(app, "on_delete_item_%s" % resource).call_async(original)
 
     if soft_delete_enabled:
         # Instead of removing the document from the db, just mark it as deleted
@@ -132,7 +132,7 @@ async def deleteitem_internal(
         # and add deleted version
         insert_versioning_documents(resource, marked_document)
         # update oplog if needed
-        oplog_push(resource, marked_document, "DELETE", id)
+        await oplog_push(resource, marked_document, "DELETE", id)
 
     else:
         # Delete the document for real
@@ -172,11 +172,11 @@ async def deleteitem_internal(
             )
 
         # update oplog if needed
-        oplog_push(resource, original, "DELETE", id)
+        await oplog_push(resource, original, "DELETE", id)
 
     if not suppress_callbacks:
-        getattr(app, "on_deleted_item")(resource, original)
-        getattr(app, "on_deleted_item_%s" % resource)(original)
+        await getattr(app, "on_deleted_item").call_async(resource, original)
+        await getattr(app, "on_deleted_item_%s" % resource).call_async(original)
 
     return all_done()
 
@@ -206,8 +206,8 @@ async def delete(resource, **lookup):
     """
 
     resource_def = config.DOMAIN[resource]
-    getattr(app, "on_delete_resource")(resource)
-    getattr(app, "on_delete_resource_%s" % resource)()
+    await getattr(app, "on_delete_resource").call_async(resource)
+    await getattr(app, "on_delete_resource_%s" % resource).call_async()
     default_request = ParsedRequest()
     if resource_def["soft_delete"]:
         # get_document should always fetch soft deleted documents from the db
@@ -218,8 +218,8 @@ async def delete(resource, **lookup):
     if not originals:
         return all_done()
     # I add new callback as I want the framework to be retro-compatible
-    getattr(app, "on_delete_resource_originals")(resource, originals, lookup)
-    getattr(app, "on_delete_resource_originals_%s" % resource)(originals, lookup)
+    await getattr(app, "on_delete_resource_originals").call_async(resource, originals, lookup)
+    await getattr(app, "on_delete_resource_originals_%s" % resource).call_async(originals, lookup)
     id_field = resource_def["id_field"]
 
     if resource_def["soft_delete"]:
@@ -250,7 +250,7 @@ async def delete(resource, **lookup):
         if resource_def["versioning"] is True:
             app.data.remove(resource + config.VERSIONS, lookup)
 
-    getattr(app, "on_deleted_resource")(resource)
-    getattr(app, "on_deleted_resource_%s" % resource)()
+    await getattr(app, "on_deleted_resource").call_async(resource)
+    await getattr(app, "on_deleted_resource_%s" % resource).call_async()
 
     return all_done()

eve/methods/get.py

@@ -285,8 +285,8 @@ async def _perform_find(resource, lookup):
     # functions modify the documents, the last_modified and etag won't be
     # updated to reflect the changes (they always reflect the documents
     # state on the database.)
-    getattr(app, "on_fetched_resource")(resource, response)
-    getattr(app, "on_fetched_resource_%s" % resource)(response)
+    await getattr(app, "on_fetched_resource").call_async(resource, response)
+    await getattr(app, "on_fetched_resource_%s" % resource).call_async(response)
 
     # the 'extra' cursor field, if present, will be added to the response.
     # Can be used by Eve extensions to add extra, custom data to any
@@ -524,15 +524,15 @@ async def getitem_internal(resource, **lookup):
             versions = response[config.ITEMS]
 
         if version == "diffs":
-            getattr(app, "on_fetched_diffs")(resource, versions)
-            getattr(app, "on_fetched_diffs_%s" % resource)(versions)
+            await getattr(app, "on_fetched_diffs").call_async(resource, versions)
+            await getattr(app, "on_fetched_diffs_%s" % resource).call_async(versions)
         else:
             for version_item in versions:
-                getattr(app, "on_fetched_item")(resource, version_item)
-                getattr(app, "on_fetched_item_%s" % resource)(version_item)
+                await getattr(app, "on_fetched_item").call_async(resource, version_item)
+                await getattr(app, "on_fetched_item_%s" % resource).call_async(version_item)
     else:
-        getattr(app, "on_fetched_item")(resource, response)
-        getattr(app, "on_fetched_item_%s" % resource)(response)
+        await getattr(app, "on_fetched_item").call_async(resource, response)
+        await getattr(app, "on_fetched_item_%s" % resource).call_async(response)
 
     return response, last_modified, etag, 200
 

eve/methods/patch.py

@@ -206,8 +206,8 @@ async def patch_internal(
             updated = deepcopy(original)
 
             # notify callbacks
-            getattr(app, "on_update")(resource, updates, original)
-            getattr(app, "on_update_%s" % resource)(updates, original)
+            await getattr(app, "on_update").call_async(resource, updates, original)
+            await getattr(app, "on_update_%s" % resource).call_async(updates, original)
 
             if resource_def["merge_nested_documents"]:
                 updates = resolve_nested_documents(updates, updated)
@@ -225,13 +225,13 @@ async def patch_internal(
                     abort(412, description="Client and server etags don't match")
 
             # update oplog if needed
-            oplog_push(resource, updates, "PATCH", object_id)
+            await oplog_push(resource, updates, "PATCH", object_id)
 
             insert_versioning_documents(resource, updated)
 
             # nofity callbacks
-            getattr(app, "on_updated")(resource, updates, original)
-            getattr(app, "on_updated_%s" % resource)(updates, original)
+            await getattr(app, "on_updated").call_async(resource, updates, original)
+            await getattr(app, "on_updated_%s" % resource).call_async(updates, original)
 
             updated.update(updates)
 

eve/methods/post.py

@@ -243,8 +243,8 @@ async def post_internal(resource, payl=None, skip_validation=False):
         return_code = config.VALIDATION_ERROR_STATUS
     else:
         # notify callbacks
-        getattr(app, "on_insert")(resource, documents)
-        getattr(app, "on_insert_%s" % resource)(documents)
+        await getattr(app, "on_insert").call_async(resource, documents)
+        await getattr(app, "on_insert_%s" % resource).call_async(documents)
 
         # compute etags here as documents might have been updated by callbacks.
         resolve_document_etag(documents, resource)
@@ -253,7 +253,7 @@ async def post_internal(resource, payl=None, skip_validation=False):
         ids = app.data.insert(resource, documents)
 
         # update oplog if needed
-        oplog_push(resource, documents, "POST")
+        await oplog_push(resource, documents, "POST")
 
         # assign document ids
         for document in documents:
@@ -277,8 +277,8 @@ async def post_internal(resource, payl=None, skip_validation=False):
         insert_versioning_documents(resource, documents)
 
         # notify callbacks
-        getattr(app, "on_inserted")(resource, documents)
-        getattr(app, "on_inserted_%s" % resource)(documents)
+        await getattr(app, "on_inserted").call_async(resource, documents)
+        await getattr(app, "on_inserted_%s" % resource).call_async(documents)
         # request was received and accepted; at least one document passed
         # validation and was accepted for insertion.
 

eve/methods/put.py

@@ -196,8 +196,8 @@ async def put_internal(
             resolve_document_version(document, resource, "PUT", original)
 
             # notify callbacks
-            getattr(app, "on_replace")(resource, document, original)
-            getattr(app, "on_replace_%s" % resource)(document, original)
+            await getattr(app, "on_replace").call_async(resource, document, original)
+            await getattr(app, "on_replace_%s" % resource).call_async(document, original)
 
             resolve_document_etag(document, resource)
 
@@ -209,13 +209,13 @@ async def put_internal(
                     abort(412, description="Client and server etags don't match")
 
             # update oplog if needed
-            oplog_push(resource, document, "PUT")
+            await oplog_push(resource, document, "PUT")
 
             insert_versioning_documents(resource, document)
 
             # notify callbacks
-            getattr(app, "on_replaced")(resource, document, original)
-            getattr(app, "on_replaced_%s" % resource)(document, original)
+            await getattr(app, "on_replaced").call_async(resource, document, original)
+            await getattr(app, "on_replaced_%s" % resource).call_async(document, original)
 
             # build the full response document
             build_response_document(document, resource, embedded_fields, document)

eve/render.py

@@ -46,17 +46,17 @@ def raise_event(f):
     """
 
     @wraps(f)
-    def decorated(*args, **kwargs):
-        r = f(*args, **kwargs)
+    async def decorated(*args, **kwargs):
+        r = await f(*args, **kwargs)
         method = request.method
         if method in ("GET", "POST", "PATCH", "DELETE", "PUT"):
             event_name = "on_post_" + method
             resource = args[0] if args else None
             # general hook
-            getattr(app, event_name)(resource, request, r)
+            await getattr(app, event_name).call_async(resource, request, r)
             if resource:
                 # resource hook
-                getattr(app, event_name + "_" + resource)(request, r)
+                await getattr(app, event_name + "_" + resource).call_async(request, r)
         return r
 
     return decorated

setup.py

@@ -14,7 +14,6 @@
 
 INSTALL_REQUIRES = [
     "cerberus>=1.1,<2.0",
-    "events>=0.3,<0.4",
     "quart @ git+https://github.com/MarkLark86/quart@fix-test-client-with-utf8-url",
     "pymongo",
     "simplejson>=3.3.0,<4.0",