Merge pull request #1 from clef/initial-logic

feat(flask_nameko): adds FlaskPooledClusterRpcProxy

Author: jessepollak

MANIFEST.in

@@ -1,6 +1,6 @@
 include HISTORY.rst
 include LICENSE
-include README.rst
+include README.md
 
 recursive-include tests *
 recursive-exclude * __pycache__

Makefile

@@ -45,10 +45,10 @@ clean-test:
 	rm -fr htmlcov/
 
 lint:
-	flake8 flask-nameko tests
+	flake8 flask_nameko tests
 
 develop: clean
-	virtualenv venv
+	virtualenv --python=python2.7 venv
 	. venv/bin/activate && pip install -r requirements_dev.txt
 
 test:
@@ -61,7 +61,7 @@ test-all:
 	tox
 
 coverage:
-	coverage run --source flask-nameko setup.py test
+	coverage run --source flask_nameko setup.py test
 	coverage report -m
 	coverage html
 	$(BROWSER) htmlcov/index.html

README.md

@@ -1,21 +1,68 @@
-# flask-nameko
+# flask_nameko
 
 A wrapper for using nameko services with Flask
 
 ## Installation
 
 Install it via Clef's PyPI:
 
-    pip install flask-nameko
+    pip install flask_nameko
 
 ## Usage
-    import flask-nameko
 
-    # Add more usage information here.
-    # Good code is bad unless it's documented.
+To start using `flask_nameko`, you need to create and configure a new `FlaskPooledClusterRpcProxy` singleton, which you'll use to communicate with your Nameko cluster.
+
+    # __init__.py
+    from flask import Flask
+    from flask_nameko import FlaskPooledClusterRpcProxy
+
+    rpc = FlaskPooledClusterRpcProxy()
+
+    def create_app():
+        app = Flask(__name__)
+        app.config.update(dict(
+            NAMEKO_AMQP_URI='amqp://localhost'
+        ))
+
+        rpc.init_app(app)
+
+    app = create_app()
+
+Then, you can use the `FlaskPooledClusterRpcProxy` singleton just as you would normally use a `ClusterRpcProxy`, by accessing individual services by name and calling methods on them:
+
+    # routes.py
+
+    from . import (
+        app,
+        rpc
+    )
+
+    @app.route('/'):
+    def index():
+        result = rpc.service.do_something('test')
+        return result
+
+## API
+
+### Configuration
+
+`FlaskPooledClusterRpcProxy` accepts all nameko configuration values, prefixed with the `NAMEKO_` prefix. In addition, it exposes additional configuration options:
+
+* `INITIAL_CONNECTIONS (int, default=2)` - the number of initial connections to the Nameko cluster to create
+* `MAX_CONNECTIONS (int, default=8)` - the max number of connections to the Nameko cluster to create before raises an error
+
+### Proxies
+
+*flask_nameko.**FlaskPooledClusterRpcProxy**(app=None)*
+
+   This class is used to create a pool of connections to a Nameko cluster.
+
+   *init_app(app=None)*
+
+   Configure the proxy for a given app.
 
 ## Development
 
-    $ git clone git@github.com:clef/flask-nameko.git flask-nameko
-    $ cd flask-nameko
+    $ git clone git@github.com:clef/flask_nameko.git flask_nameko
+    $ cd flask_nameko
     $ make develop

circle.yml

@@ -1,30 +1,3 @@
-machine:
-  environment:
-    PIP_CERT: /home/ubuntu/.pypi.pem
-    PIP_INTERNAL_CERT: |
-      -----BEGIN CERTIFICATE-----
-      MIIDVDCCAjwCCQDgHdRtPoGIUDANBgkqhkiG9w0BAQsFADBsMQswCQYDVQQGEwJV
-      UzETMBEGA1UECBMKQ2FsaWZvcm5pYTENMAsGA1UEChMEQ2xlZjE5MDcGA1UEAxMw
-      Y2xlZi1weXBpLTE3ODA3NjY5MjUudXMtZWFzdC0xLmVsYi5hbWF6b25hd3MuY29t
-      MB4XDTE2MDQwODIzMjkyMVoXDTE3MDQwODIzMjkyMVowbDELMAkGA1UEBhMCVVMx
-      EzARBgNVBAgTCkNhbGlmb3JuaWExDTALBgNVBAoTBENsZWYxOTA3BgNVBAMTMGNs
-      ZWYtcHlwaS0xNzgwNzY2OTI1LnVzLWVhc3QtMS5lbGIuYW1hem9uYXdzLmNvbTCC
-      ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAJ3vXy95gtyKgvsLI9G9H69s
-      bFxnhdCbbQWLTFUypLWN92qssP73UZUHQsg7NvGLvUK6r99BtojsomC6KGqrt37S
-      6gLQ0FnGmaMeGZw2f8T4Gdn+txegbZryNttxsEOX4uD7J9V818x/zv/1USheD/qV
-      MrQENRvX0BbveJN7hIbTLN1+IQlSSoxLXSqf/gshUK99jhKSOaqxwSuweEuRInn8
-      2jFh6GwLJL3mfpGYv6PnK+io3hgxljcC329ae4N/iegg3UawzGTJNT5AGpRjBAos
-      vRERpu2wjqZ+y0YsWya1YSYwoleq7UYV9xXRAlavWy0XmSwXNGdHOXkiQjUUvmkC
-      AwEAATANBgkqhkiG9w0BAQsFAAOCAQEAITaAnchir51dItbTAkKoJbNxS3kUmh57
-      Y168Z+MblOoRPxjpeue+QXwErVMAALqAKgC9vOofDThSO0Z8jcKRG5OXXbH0fbIz
-      WqYznjrbycAP3ecrcNtIB3gLPHCYabRHto/j0Zcn+7QwAmAVBNGvtRgldOL9TOm7
-      JHrRDF3/thMtdRQfhbh1S1dBP+uuH+ouGfOJgP0LN5zu35vSKmKpLrSerTvW9yOK
-      q4vttoEKuILwTc2ED94T9aw7Fy8zzh9477lb0fQUbh/aSF5oKX+SYQP0B0rB9qyJ
-      klFCOKVEmTqvPdUG8VfEd+YO9EQI48S4HLJwwgnWIfV5lVjGgwpJ+w==
-      -----END CERTIFICATE-----
-  post:
-    # Set up pip to use our private PyPI server
-    - echo -e "$PIP_INTERNAL_CERT" | cat - `python -m pip._vendor.requests.certs` > $PIP_CERT
 dependencies:
   cache_directories:
     - .tox

flask-nameko/flask-nameko.py

@@ -3,3 +3,6 @@
 __author__ = 'Jesse Pollak'
 __email__ = 'jesse@getclef.com'
 __version__ = '0.1.0'
+
+from errors import *
+from proxies import FlaskPooledClusterRpcProxy

flask-nameko/__init__.py renamed to flask_nameko/__init__.py

@@ -0,0 +1,65 @@
+from Queue import Queue, Empty
+from threading import Lock
+from .errors import ClientUnavailableError
+
+class ConnectionPool(object):
+    def __init__(self, get_connection, initial_connections=2, max_connections=8):
+        """
+        Create a new pool
+        :param func get_connection: The function that returns a connection
+        :param int initial_connections: The initial number of connection objects to create
+        :param int max_connections: The maximum amount of connections to create. These
+          connections will only be created on demand and will potentially be
+          destroyed once they have been returned via a call to
+          :meth:`release_connection`
+        constructor
+        """
+        self._get_connection = get_connection
+        self._queue = Queue()
+        self._current_connections = 0
+        self._max_connections = max_connections
+        self._lock = Lock()
+
+        for x in range(initial_connections):
+            connection = self._make_connection()
+            self._queue.put(connection)
+
+    def _make_connection(self):
+        ret = self._get_connection()
+        self._current_connections += 1
+        return ret
+
+    def get_connection(self, initial_timeout=0.05, next_timeout=1):
+        """
+        Wait until a connection instance is available
+        :param float initial_timeout:
+          how long to wait initially for an existing connection to complete
+        :param float next_timeout:
+          if the pool could not obtain a connection during the initial timeout,
+          and we have allocated the maximum available number of connections, wait
+          this long until we can retrieve another one
+        :return: A connection object
+        """
+        try:
+            return self._queue.get(True, initial_timeout)
+        except Empty:
+            try:
+                self._lock.acquire()
+                if self._current_connections == self._max_connections:
+                    raise ClientUnavailableError("Too many connections in use")
+                cb = self._make_connection()
+                return cb
+            except ClientUnavailableError as ex:
+                try:
+                    return self._queue.get(True, next_timeout)
+                except Empty:
+                    raise ex
+            finally:
+                self._lock.release()
+
+    def release_connection(self, cb):
+        """
+        Return a Connection object to the pool
+        :param Connection cb: the connection to release
+        """
+        self._queue.put(cb, True)

flask_nameko/connection_pool.py

@@ -0,0 +1,4 @@
+class BadConfigurationError(Exception):
+    pass
+class ClientUnavailableError(Exception):
+    pass

flask_nameko/errors.py

@@ -0,0 +1,73 @@
+import re
+from flask import g
+from nameko.standalone.rpc import ClusterRpcProxy
+from .connection_pool import ConnectionPool
+from .errors import (
+    BadConfigurationError
+)
+
+class PooledClusterRpcProxy(object):
+
+    _pool = None
+    _config = None
+
+    def __init__(self, config=None):
+        if config:
+            self.configure(config)
+
+    def configure(self, config):
+        if not config.get('AMQP_URI'):
+            raise BadConfigurationError("Please provide a valid configuration.")
+
+        self._config = config
+        self._pool = ConnectionPool(
+            self._get_nameko_connection,
+            initial_connections=config.get('INITIAL_CONNECTIONS', 2),
+            max_connections=config.get('MAX_CONNETIONS', 8)
+        )
+
+    def _get_nameko_connection(self):
+        proxy = ClusterRpcProxy(self._config)
+        return proxy.start()
+
+    def get_connection(self):
+        if not self._pool:
+            raise ClusterNotConfiguredError("Please configure your cluster beore requesting a connection.")
+        return self._pool.get_connection()
+
+    def release_connection(self, connection):
+        return self._pool.release_connection(connection)
+
+
+class FlaskPooledClusterRpcProxy(PooledClusterRpcProxy):
+    def __init__(self, app=None):
+        if app:
+            self.init_app(app)
+
+    def init_app(self, app):
+        config = dict()
+        for key, val in app.config.iteritems():
+            match = re.match(r"NAMEKO\_(?P<name>.*)", key)
+            if match:
+                config[match.group('name')] = val
+        self.configure(config)
+        app.teardown_appcontext(self._teardown_nameko_connection)
+
+    def get_connection(self):
+        connection = getattr(g, '_nameko_connection', None)
+        if not connection:
+            connection = super(FlaskPooledClusterRpcProxy, self).get_connection()
+            g._nameko_connection = connection
+        return connection
+
+    def _teardown_nameko_connection(self, exception):
+        connection = getattr(g, '_nameko_connection', None)
+        if connection is not None:
+            self.release_connection(connection)
+
+    def __getattr__(self, name):
+        return getattr(self.get_connection(), name)
+
+    def __getitem__(self, name):
+        return getattr(self.get_connection(), name)
+

flask_nameko/proxies.py

@@ -1,2 +1,3 @@
 [pytest]
 norecursedirs = venv .git *.egg *site-packages*
+addopts = -s