Add support for signals via blinker (#278)

Author: Lita Cho

docs/signals.rst

@@ -0,0 +1,48 @@
+Signals
+=======
+Starting with PynamoDB 3.1.0, there is support for signalling.  This support is provided by the `blinker`_ library, which is not installed by default. In order to ensure blinker is installed, specify your PynamoDB requirement like so:
+
+::
+
+	pynamodb[signals]==<YOUR VERSION NUMBER>
+
+Signals allow certain senders to notify subscribers that something happened. PynamoDB currently sends signals before and after every DynamoDB API call.
+
+.. note::
+
+    It is recommended to avoid business logic in signal callbacks, as this can have performance implications. To reinforce this, only the operation name and table name are available in the signal callback.
+
+
+Subscribing to Signals
+----------------------
+
+PynamoDB fires two signal calls, `pre_dynamodb_send` before the network call and `post_dynamodb_send` after the network call to DynamoDB.
+
+The callback must taking the following arguments:
+
+================  ===========
+Arguments         Description
+================  ===========
+*sender*          The object that fired that method.
+*operation_name*  The string name of the `DynamoDB action`_
+*table_name*      The name of the table the operation is called upon.
+*req_uuid*        A unique identifer so subscribers can correlate the before and after events.
+================  ===========
+
+To subscribe to a signal, the user needs to import the signal object and connect your callback, like so.
+
+.. code:: python
+
+    from pynamodb.signals import pre_dynamodb_send, post_dynamodb_send
+
+    def record_pre_dynamodb_send(sender, operation_name, table_name, req_uuid):
+        pre_recorded.append((operation_name, table_name, req_uuid))
+
+    def record_post_dynamodb_send(sender, operation_name, table_name, req_uuid):
+        post_recorded.append((operation_name, table_name, req_uuid))
+
+    pre_dynamodb_send.connect(record_pre_dynamodb_send)
+    post_dynamodb_send.connect(record_post_dynamodb_send)
+
+.. _blinker: https://pypi.python.org/pypi/blinker
+.. _Dynamo action: https://github.com/pynamodb/PynamoDB/blob/cd705cc4e0e3dd365c7e0773f6bc02fe071a0631/

pynamodb/connection/base.py

@@ -2,26 +2,24 @@
 Lowest level connection
 """
 from __future__ import division
-from base64 import b64decode
+
 import logging
 import math
 import random
 import time
+import uuid
+from base64 import b64decode
 
 import six
-from six.moves import range
-from botocore.session import get_session
-from botocore.exceptions import BotoCoreError
 from botocore.client import ClientError
+from botocore.exceptions import BotoCoreError
+from botocore.session import get_session
 from botocore.vendored import requests
+from botocore.vendored.requests import Request
+from six.moves import range
 
-from pynamodb.connection.util import pythonic
-from pynamodb.types import HASH, RANGE
 from pynamodb.compat import NullHandler
-from pynamodb.exceptions import (
-    TableError, QueryError, PutError, DeleteError, UpdateError, GetError, ScanError, TableDoesNotExist,
-    VerboseClientError
-)
+from pynamodb.connection.util import pythonic
 from pynamodb.constants import (
     RETURN_CONSUMED_CAPACITY_VALUES, RETURN_ITEM_COLL_METRICS_VALUES, COMPARISON_OPERATOR_VALUES,
     RETURN_ITEM_COLL_METRICS, RETURN_CONSUMED_CAPACITY, RETURN_VALUES_VALUES, ATTR_UPDATE_ACTIONS,
@@ -38,8 +36,13 @@
     CONDITIONAL_OPERATORS, NULL, NOT_NULL, SHORT_ATTR_TYPES, DELETE,
     ITEMS, DEFAULT_ENCODING, BINARY_SHORT, BINARY_SET_SHORT, LAST_EVALUATED_KEY, RESPONSES, UNPROCESSED_KEYS,
     UNPROCESSED_ITEMS, STREAM_SPECIFICATION, STREAM_VIEW_TYPE, STREAM_ENABLED)
+from pynamodb.exceptions import (
+    TableError, QueryError, PutError, DeleteError, UpdateError, GetError, ScanError, TableDoesNotExist,
+    VerboseClientError
+)
 from pynamodb.settings import get_settings_value
-from botocore.vendored.requests import Request
+from pynamodb.signals import pre_dynamodb_send, post_dynamodb_send
+from pynamodb.types import HASH, RANGE
 
 BOTOCORE_EXCEPTIONS = (BotoCoreError, ClientError)
 
@@ -261,7 +264,12 @@ def dispatch(self, operation_name, operation_kwargs):
                 operation_kwargs.update(self.get_consumed_capacity_map(TOTAL))
         self._log_debug(operation_name, operation_kwargs)
 
+        table_name = operation_kwargs.get(TABLE_NAME)
+        req_uuid = uuid.uuid4()
+
+        self.send_pre_boto_callback(operation_name, req_uuid, table_name)
         data = self._make_api_call(operation_name, operation_kwargs)
+        self.send_post_boto_callback(operation_name, req_uuid, table_name)
 
         if data and CONSUMED_CAPACITY in data:
             capacity = data.get(CONSUMED_CAPACITY)
@@ -270,6 +278,18 @@ def dispatch(self, operation_name, operation_kwargs):
             log.debug("%s %s consumed %s units",  data.get(TABLE_NAME, ''), operation_name, capacity)
         return data
 
+    def send_post_boto_callback(self, operation_name, req_uuid, table_name):
+        try:
+            post_dynamodb_send.send(self, operation_name=operation_name, table_name=table_name, req_uuid=req_uuid)
+        except Exception as e:
+            log.exception("post_boto callback threw an exception.")
+
+    def send_pre_boto_callback(self, operation_name, req_uuid, table_name):
+        try:
+            pre_dynamodb_send.send(self, operation_name=operation_name, table_name=table_name, req_uuid=req_uuid)
+        except Exception as e:
+            log.exception("pre_boto callback threw an exception.")
+
     def _make_api_call(self, operation_name, operation_kwargs):
         """
         This private method is here for two reasons:
@@ -281,7 +301,6 @@ def _make_api_call(self, operation_name, operation_kwargs):
             operation_kwargs,
             operation_model
         )
-
         prepared_request = self._create_prepared_request(request_dict, operation_model)
 
         for i in range(0, self._max_retry_attempts_exception + 1):

pynamodb/signals.py

@@ -0,0 +1,50 @@
+"""
+Implements signals based on blinker if available, otherwise
+falls silently back to a noop.
+
+This implementation was taken from Flask:
+https://github.com/pallets/flask/blob/master/flask/signals.py
+"""
+signals_available = False
+
+
+class _FakeNamespace(object):
+    def signal(self, name, doc=None):
+        return _FakeSignal(name, doc)
+
+
+class _FakeSignal(object):
+    """
+    If blinker is unavailable, create a fake class with the same
+    interface that allows sending of signals but will fail with an
+    error on anything else.  Instead of doing anything on send, it
+    will just ignore the arguments and do nothing instead.
+    """
+
+    def __init__(self, name, doc=None):
+        self.name = name
+        self.__doc__ = doc
+
+    def _fail(self, *args, **kwargs):
+        raise RuntimeError('signalling support is unavailable '
+                           'because the blinker library is '
+                           'not installed.')
+
+    send = lambda *a, **kw: None  # noqa
+    connect = disconnect = has_receivers_for = receivers_for = \
+        temporarily_connected_to = _fail
+    del _fail
+
+
+try:
+    from blinker import Namespace
+    signals_available = True
+except ImportError: # pragma: no cover
+    Namespace = _FakeNamespace
+
+# The namespace for code signals.  If you are not PynamoDB code, do
+# not put signals in here.  Create your own namespace instead.
+_signals = Namespace()
+
+pre_dynamodb_send = _signals.signal('pre_dynamodb_send')
+post_dynamodb_send = _signals.signal('post_dynamodb_send')

pynamodb/tests/test_signals.py

@@ -0,0 +1,98 @@
+import mock
+import pytest
+
+from pynamodb.connection import Connection
+from pynamodb.signals import _FakeNamespace
+from pynamodb.signals import pre_dynamodb_send, post_dynamodb_send
+
+try:
+    import blinker
+except ImportError:
+    blinker = None
+
+PATCH_METHOD = 'pynamodb.connection.Connection._make_api_call'
+
+
+@mock.patch(PATCH_METHOD)
+@mock.patch('pynamodb.connection.base.uuid')
+def test_signal(mock_uuid, mock_req):
+    pre_recorded = []
+    post_recorded = []
+    UUID = '123-abc'
+
+    def record_pre_dynamodb_send(sender, operation_name, table_name, req_uuid):
+        pre_recorded.append((operation_name, table_name, req_uuid))
+
+    def record_post_dynamodb_send(sender, operation_name, table_name, req_uuid):
+        post_recorded.append((operation_name, table_name, req_uuid))
+
+    pre_dynamodb_send.connect(record_pre_dynamodb_send)
+    post_dynamodb_send.connect(record_post_dynamodb_send)
+    try:
+        mock_uuid.uuid4.return_value = UUID
+        mock_req.return_value = {'TableDescription': {'TableName': 'table', 'TableStatus': 'Creating'}}
+        c = Connection()
+        c.dispatch('CreateTable', {'TableName': 'MyTable'})
+        assert ('CreateTable', 'MyTable', UUID) == pre_recorded[0]
+        assert ('CreateTable', 'MyTable', UUID) == post_recorded[0]
+    finally:
+        pre_dynamodb_send.disconnect(record_pre_dynamodb_send)
+        post_dynamodb_send.disconnect(record_post_dynamodb_send)
+
+
+@mock.patch(PATCH_METHOD)
+@mock.patch('pynamodb.connection.base.uuid')
+def test_signal_exception_pre_signal(mock_uuid, mock_req):
+    post_recorded = []
+    UUID = '123-abc'
+
+    def record_pre_dynamodb_send(sender, operation_name, table_name, req_uuid):
+        raise ValueError()
+
+    def record_post_dynamodb_send(sender, operation_name, table_name, req_uuid):
+        post_recorded.append((operation_name, table_name, req_uuid))
+
+    pre_dynamodb_send.connect(record_pre_dynamodb_send)
+    post_dynamodb_send.connect(record_post_dynamodb_send)
+    try:
+        mock_uuid.uuid4.return_value = UUID
+        mock_req.return_value = {'TableDescription': {'TableName': 'table', 'TableStatus': 'Creating'}}
+        c = Connection()
+        c.dispatch('CreateTable', {'TableName': 'MyTable'})
+        assert ('CreateTable', 'MyTable', UUID) == post_recorded[0]
+    finally:
+        pre_dynamodb_send.disconnect(record_pre_dynamodb_send)
+        post_dynamodb_send.disconnect(record_post_dynamodb_send)
+
+
+@mock.patch(PATCH_METHOD)
+@mock.patch('pynamodb.connection.base.uuid')
+def test_signal_exception_post_signal(mock_uuid, mock_req):
+    pre_recorded = []
+    UUID = '123-abc'
+
+    def record_pre_dynamodb_send(sender, operation_name, table_name, req_uuid):
+        pre_recorded.append((operation_name, table_name, req_uuid))
+
+    def record_post_dynamodb_send(sender, operation_name, table_name, req_uuid):
+        raise ValueError()
+
+    pre_dynamodb_send.connect(record_pre_dynamodb_send)
+    post_dynamodb_send.connect(record_post_dynamodb_send)
+    try:
+        mock_uuid.uuid4.return_value = UUID
+        mock_req.return_value = {'TableDescription': {'TableName': 'table', 'TableStatus': 'Creating'}}
+        c = Connection()
+        c.dispatch('CreateTable', {'TableName': 'MyTable'})
+        assert ('CreateTable', 'MyTable', UUID) == pre_recorded[0]
+    finally:
+        pre_dynamodb_send.disconnect(record_pre_dynamodb_send)
+        post_dynamodb_send.disconnect(record_post_dynamodb_send)
+
+
+def test_fake_signals():
+    _signals = _FakeNamespace()
+    pre_dynamodb_send = _signals.signal('pre_dynamodb_send')
+    with pytest.raises(RuntimeError):
+        pre_dynamodb_send.connect(lambda x: x)
+    pre_dynamodb_send.send(object, operation_name="UPDATE", table_name="TEST", req_uuid="something")

requirements-dev.txt

@@ -6,6 +6,7 @@ pytest-cov==2.4.0
 pytest-mock==1.6.0
 python-coveralls==2.5.0
 tox==2.1.1
+blinker==1.4
 flake8==2.4.1
 requests==2.7.0
 # Pinned because latest version requires Python 2.7+ (https://github.com/sphinx-doc/sphinx/issues/2919).

setup.py

@@ -39,4 +39,7 @@
         'Programming Language :: Python :: 3.3',
         'License :: OSI Approved :: MIT License',
     ],
+    extras_require={
+        'signals': ['blinker>=1.3,<2.0'] 
+    },
 )