Merge pull request #36 from linewalks/develop

update dasima 0.2.0 (2022-01-12)

.gitignore

@@ -1,3 +1,8 @@
+
+# Created by https://www.toptal.com/developers/gitignore/api/visualstudiocode,python
+# Edit at https://www.toptal.com/developers/gitignore?templates=visualstudiocode,python
+
+### Python ###
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]
@@ -20,7 +25,6 @@ parts/
 sdist/
 var/
 wheels/
-pip-wheel-metadata/
 share/python-wheels/
 *.egg-info/
 .installed.cfg
@@ -50,6 +54,7 @@ coverage.xml
 *.py,cover
 .hypothesis/
 .pytest_cache/
+cover/
 
 # Translations
 *.mo
@@ -72,6 +77,7 @@ instance/
 docs/_build/
 
 # PyBuilder
+.pybuilder/
 target/
 
 # Jupyter Notebook
@@ -82,7 +88,9 @@ profile_default/
 ipython_config.py
 
 # pyenv
-.python-version
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
 
 # pipenv
 #   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
@@ -127,3 +135,28 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+### VisualStudioCode ###
+.vscode
+.vscode/*
+!.vscode/settings.json
+!.vscode/tasks.json
+!.vscode/launch.json
+!.vscode/extensions.json
+*.code-workspace
+
+# Local History for Visual Studio Code
+.history/
+
+### VisualStudioCode Patch ###
+# Ignore all local history of files
+.history
+.ionide
+
+# End of https://www.toptal.com/developers/gitignore/api/visualstudiocode,python

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2021 linewalks
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,2 +1,119 @@
-# CLUE-MQ
-CLUE pub-sub을 위한 repo
+# DASIMA
+
+#### Description
+
+Dasima is a library that helps you send and receive messages in Flask project. It is a simple wrapper around Kombu and support with the publisher/subscriber pattern of your Flask project.
+
+
+
+
+## Getting Started
+
+#### Built With
+
+* Python >=  3.6
+
+
+
+#### Prerequisites
+
+- Message Queue 
+
+  > ex) Redis, RabbitMQ, ActiveMQ, ZeroMQ, Kafka...
+
+
+
+#### Installation
+
+##### Installing
+
+```shell
+$ pip install dasima
+```
+
+
+
+##### setting parameters
+
+```python
+DASIMA_CONNECTION_HOST = "localhost" # your Message Queue host ex) redis://0.0.0.0, amqp://id:password@0.0.0.0:port
+DASIMA_ACCEPT_TYPE = "json" # sending data type ex) json, pickle ...
+DASIMA_EXCHANGE_SETTING = [("test_exchange", "one"),] 
+# DASIMA_EXCHANGE_SETTING is list of tuples [(exchange name, type)]
+# There are two types: 'all' and 'one'.
+```
+
+
+
+## Usage
+
+#### Subscriber Simple example
+
+```python
+from dasima import Dasima
+from flask import Flask
+
+
+app = Flask(__name__)
+
+dasimamq = Dasima()
+dasimamq.init_app(app) # Alternatively, auto init_app can be used after putting the flask app into Dasima like Dasima(app).
+
+
+# Be able to subscribe target functions using the function 'subscribe' 
+# The queue named by subscribed function name will be made, and binding it with routing key
+# dasimamq.{exchange}.subscribe(routing_key) - "Route key to bind"
+@dasimamq.test_exchange.subscribe(routing_key="test_routing_key")
+# @dasimamq.test_exchange.subscribe - if routing key not defined, routing key is defined as function name
+def test_function(x, y):
+    print(x + y)
+    return x + y
+
+
+if __name__ == "__main__":
+ # Call the function 'run_subscribers' to create queues in which consumers process the messages.
+    dasimamq.run_subscribers()
+    app.run(port=5050)
+```
+
+
+
+#### Publisher Simple example
+
+```python
+from flask import Flask
+from dasima import Dasima
+
+
+app = Flask(__name__)
+
+
+dasimamq = Dasima()
+dasimamq.init_app(app) # Alternatively, auto init_app is possible by putting the flask app directly into Dasima(app).
+
+
+@app.route("/")
+def send_message():
+    # dasimamq.{exchange}.subscribe("Route key to bind")
+    dasimamq.test_exchange.send_message({"x": 1, "y": 2}, "test_routing_key")
+    return {"data": "send message successful"}
+
+
+if __name__ == "__main__":
+    app.run(port=5000)
+```
+
+
+
+## Links
+
+- [Kombu github](https://github.com/celery/kombu)
+- [Redis](https://redis.io/)
+- [RabbitMQ](https://www.rabbitmq.com/)
+
+
+
+## Contact
+
+**JISU JEONG** - js.jeong@linewalks.com
+

dasima/__init__.py

@@ -0,0 +1,67 @@
+import threading
+
+from kombu import Connection
+
+from dasima.exchange import ExchangeWrapper
+from dasima.worker import Worker
+
+
+class Dasima:
+  def __init__(self, app=None):
+    self.app = app
+    if app:
+      self.init_app(app)
+
+  def init_app(self, app):
+    self.app = app
+    self.app_ctx = self.app.app_context()
+    self.exchange_list = self.app.config.get(
+        "DASIMA_EXCHANGE_SETTING",
+        [("dasima_test", "one")]
+    )
+    self.connection = Connection(self.app.config.get("DASIMA_CONNECTION_HOST", "localhost"))
+    self.connection.ensure_connection(max_retries=3)
+    self.worker = Worker(
+        connection=self.connection,
+        accept_type=self.app.config.get("DASIMA_ACCEPT_TYPE", "json"),
+        app_ctx=self.app_ctx
+    )
+    self.create_exchange()
+    self.is_running = False
+
+  def create_exchange(self):
+    for exchange_name, exchange_type in self.exchange_list:
+      setattr(
+          self,
+          exchange_name,
+          ExchangeWrapper(
+              exchange_name,
+              exchange_type,
+              self.worker
+          )
+      )
+
+  def setup_queue(self):
+    for exchange_name, exchange_type in self.exchange_list:
+      exchange = getattr(
+          self,
+          exchange_name
+      )
+      self.worker.add_consumer_config_list(exchange)
+
+  def run_subscribers(self):
+    if self.is_running:
+      raise RuntimeError("run_subscribers is aleady running!")
+    else:
+      self.is_running = True
+      self.setup_queue()
+      t = threading.Thread(target=self.worker.run)
+      t.daemon = True
+      t.start()
+      # worker가 준비가 될때 까지 잠시 기다려줌
+      while True:
+        if self.worker.is_ready:
+          break
+
+  def stop_subscribers(self):
+    self.worker.stop()

dasima/exchange.py

@@ -0,0 +1,53 @@
+import uuid
+
+from kombu import Exchange, Queue, binding
+from flask.ctx import AppContext
+from dasima.worker import Worker
+
+
+class ExchangeWrapper:
+  def __init__(
+      self,
+      exchange_name: str,
+      exchange_type: str,
+      worker: Worker
+  ):
+    self.exchange_type = exchange_type
+    self.exchange = Exchange(
+        name=exchange_name,
+        type="topic",
+        durable=True
+    )
+    self.__binding_dict = {}
+    self.worker = worker
+
+  def get_binding_dict(self):
+    return self.__binding_dict
+
+  def send_message(self, data, routing_key):
+    self.worker.publish(
+        data,
+        self.exchange,
+        routing_key
+    )
+
+  def subscribe(self, routing_key=None):
+    if callable(routing_key):
+      self.add_binding_dict(routing_key, None)
+      return routing_key
+
+    def decorator(func):
+      self.add_binding_dict(func, routing_key)
+      return func
+
+    return decorator
+
+  def add_binding_dict(self, func, routing_key):
+    prefix = str(uuid.uuid4()) if self.exchange_type == "all" else ""
+    key = prefix + self.exchange.name
+    routing_key = func.__name__ if routing_key is None else routing_key
+
+    if self.__binding_dict.get(key):
+      self.__binding_dict[key].append((routing_key, func))
+    else:
+      self.__binding_dict[key] = [(routing_key, func)]

dasima/tests/conftest.py

@@ -0,0 +1,17 @@
+import pytest
+
+
+@pytest.fixture(scope="session")
+def exchange_setting_list():
+  return [
+      ("exchange_type_one", "one"),
+      ("exchange_type_all", "all")
+  ]
+
+
+@pytest.fixture(scope="session")
+def flask_app(exchange_setting_list):
+  from flask import Flask
+  app = Flask(__name__)
+  app.config["DASIMA_EXCHANGE_SETTING"] = exchange_setting_list
+  return app