Initial commit for the project

Author: Relrin

README.md

@@ -0,0 +1,41 @@
+sage-utils-python
+####################
+SDK for Open Matchmaking microservices in Python
+
+Features
+========
+- Framework agnostic solution
+- Easy to write a new extension and use it later with your code
+- Base class for implementing AMQP workers
+
+Installation
+============
+This package should be installed using pip: ::
+
+    pip install sage-utils
+
+Example
+=======
+.. code-block:: python
+
+    from sanic import Sanic
+    from sage_utils.extension import BaseExtension
+
+
+    class CustomExtension(BaseExtension):
+        extension_name = app_attribute = 'custom'
+
+        def hello(self, user):
+            print("Hello, {}!".format(user))
+
+
+    app = Sanic(__name__)
+    CustomExtension()  # available via `app.custom` or `app.extensions['custom']`
+    app.custom.hello('world')  # Hello, world!
+
+License
+=======
+The sage-utils-python is published under BSD license. For more details read LICENSE_ file.
+
+.. _links:
+.. _LICENSE: https://github.com/OpenMatchmaking/sage-utils-python/blob/master/LICENSE

README.rst

@@ -0,0 +1,4 @@
+__title__ = 'sage-utils'
+__version__ = '0.1.0'
+__license__ = 'BSD'
+VERSION = __version__

sage_utils/__init__.py

@@ -0,0 +1,3 @@
+from sage_utils.amqp.base import AmqpWorker  # NOQA
+from sage_utils.amqp.extension import AmqpExtension  # NOQA
+from sage_utils.amqp.clients import RpcAmqpClient  # NOQA

sage_utils/amqp/__init__.py

@@ -0,0 +1,28 @@
+
+
+class AmqpWorker(object):
+
+    def __init__(self, app, *args, **kwargs):
+        self.app = app
+        self.protocol = None
+        self.transport = None
+
+    async def run(self, *args, **kwargs):
+        raise NotImplementedError('`run(*args, **kwargs)` method must be implemented.')
+
+    async def connect(self):
+        self.transport, self.protocol = await self.app.amqp.connect()
+        return self.transport, self.protocol
+
+    async def free_resources(self):
+        if self.protocol:
+            if not self.protocol.worker.cancelled():
+                self.protocol.worker.cancel()
+
+            await self.protocol.close()
+
+        if self.transport:
+            self.transport.close()
+
+        self.transport = None
+        self.protocol = None

sage_utils/amqp/base.py

@@ -0,0 +1,87 @@
+import asyncio
+import json
+from copy import deepcopy
+
+
+class RpcAmqpClient(object):
+    CONTENT_TYPE = 'application/json'
+    DEFAULT_PROPERTIES = {
+        'content_type': CONTENT_TYPE,
+        'delivery_mode': 2,
+        'correlation_id': 'test-event-name'
+    }
+
+    def __init__(self, app, routing_key, request_exchange='',
+                 response_queue=None, response_exchange=''):
+        self.app = app
+        self.routing_key = routing_key
+        self.request_exchange = request_exchange
+        self.response_queue = response_queue
+        self.response_exchange = response_exchange
+        self.transport = None
+        self.protocol = None
+        self.channel = None
+
+        self.waiter = asyncio.Event()
+        self._response_queue_name = None
+        self._response = None
+
+    @property
+    def response_queue_name(self):
+        return self._response_queue_name
+
+    async def connect(self):
+        self.transport, self.protocol = await self.app.amqp.connect()
+        self.channel = await self.protocol.channel()
+
+        if self.response_queue is not None:
+            result = await self.channel.queue_declare(
+                queue_name=self.response_queue,
+                exclusive=True,
+                durable=True,
+                passive=False,
+                auto_delete=True,
+            )
+            self._response_queue_name = result['queue']
+            await self.channel.queue_bind(
+                queue_name=self.response_queue_name,
+                exchange_name=self.response_exchange,
+                routing_key=self.response_queue_name
+            )
+            await self.channel.basic_qos(
+                prefetch_count=1,
+                prefetch_size=0,
+                connection_global=False
+            )
+            await self.channel.basic_consume(
+                self.on_response,
+                queue_name=self._response_queue_name,
+            )
+
+    async def on_response(self, _channel, body, _envelope, _properties):
+        self._response = json.loads(body)
+        self.waiter.set()
+
+    async def send(self, payload={}, properties={}, raw_data=False):
+        if not self.protocol:
+            await self.connect()
+
+        request_properties = deepcopy(self.DEFAULT_PROPERTIES)
+        request_properties.update({'reply_to': self.response_queue_name})
+        request_properties.update(properties)
+        await self.channel.publish(
+            payload if raw_data else json.dumps(payload),
+            exchange_name=self.request_exchange,
+            routing_key=self.routing_key,
+            properties=request_properties
+        )
+
+        response = None
+        if self.response_queue_name is not None:
+            await self.waiter.wait()
+            response = self._response
+
+        await self.protocol.close()
+        self.protocol = None
+        self.transport = None
+        return response

sage_utils/amqp/clients.py

@@ -0,0 +1,50 @@
+from asyncio import ensure_future
+
+from aioamqp import connect as amqp_connect
+
+from sage_utils.extension import BaseExtension
+
+
+class AmqpExtension(BaseExtension):
+    app_attribute = 'amqp'
+    workers = []
+    active_tasks = []
+
+    def register_worker(self, worker):
+        self.workers.append(worker)
+
+    def get_config(self, app):
+        return {
+            "login": app.config.get("AMQP_USERNAME", "guest"),
+            "password": app.config.get("AMQP_PASSWORD", "guest"),
+            "host": app.config.get("AMQP_HOST", "localhost"),
+            "port": app.config.get("AMQP_PORT", 5672),
+            "virtualhost": app.config.get("AMQP_VIRTUAL_HOST", "vhost"),
+            "ssl": app.config.get("AMQP_USING_SSL", False),
+        }
+
+    async def connect(self):
+        config = self.get_config(self.app)
+        transport, protocol = await amqp_connect(**config)
+        return transport, protocol
+
+    async def init(self, loop):
+        if not hasattr(self.app, 'extensions'):
+            setattr(self.app, 'extensions', {})
+        self.app.extensions[self.extension_name] = self
+
+        for worker in self.workers:
+            task = ensure_future(worker.run(), loop=loop)
+            self.active_tasks.append(task)
+
+    async def deinit(self, loop):
+        for task in self.active_tasks:
+            if not loop.is_closed and not task.cancelled():
+                task.cancel()
+
+        for worker in self.workers:
+            await worker.free_resources()
+
+        setattr(self.app, self.app_attribute, None)
+        extensions = getattr(self.app, 'extensions', {})
+        extensions.pop(self.extension_name, None)

sage_utils/amqp/extension.py

@@ -0,0 +1,8 @@
+"""
+Descriptive error types, for code readability.
+"""
+AUTHORIZATION_ERROR = "AuthorizationError"
+NOT_FOUND_ERROR = "NotFoundError"
+TOKEN_ERROR = "TokenError"
+HEADER_ERROR = "HeaderError"
+VALIDATION_ERROR = "ValidationError"

sage_utils/constants.py

@@ -0,0 +1,31 @@
+
+
+class BaseExtension(object):
+    extension_name = None
+    app_attribute = None
+
+    def __init__(self, app=None, app_attribute: str=None, *args, **kwargs):
+        self.app = app
+        self.app_attribute = app_attribute or self.app_attribute
+
+        if app:
+            self._register_extension(app, *args, **kwargs)
+            self.init_app(app, *args, **kwargs)
+
+    def _register_extension(self, app, *args, **kwargs):
+        if not hasattr(app, 'extensions'):
+            setattr(app, 'extensions', {})
+
+        app.extensions[self.extension_name] = self
+
+    def init_app(self, app, *args, **kwargs):
+        setattr(app, self.app_attribute, self)
+
+    def get_from_app_config(self, app, parameter, default=None):
+        return getattr(app.config, parameter, default)
+
+    async def init(self, loop):
+        pass
+
+    async def deinit(self, loop):
+        pass

sage_utils/extension.py

@@ -0,0 +1,30 @@
+
+
+class Response(object):
+    """
+    The response object that is used by default for microservices.
+    """
+    CONTENT_FIELD_NAME = 'content'
+    ERROR_FIELD_NAME = 'error'
+    EVENT_FIELD_NAME = 'event-name'
+
+    def append_extra_fields(self, event_name, *args, **kwargs):
+        return {self.EVENT_FIELD_NAME: event_name}
+
+    def from_error(self, error_type, message, event_name=None):
+        if isinstance(message, str) and not message.endswith('.'):
+            message = message + '.'
+
+        response = {
+            self.ERROR_FIELD_NAME: {
+                "type": error_type,
+                "message": message
+            }
+        }
+        response.update(self.append_extra_fields(event_name))
+        return response
+
+    def wrap_content(self, data, event_name=None):
+        response = {self.CONTENT_FIELD_NAME: data}
+        response.update(self.append_extra_fields(event_name))
+        return response