Merge pull request #8 from RHDZMOTA/DEV

2022-09-06 RELEASE-PRD (1)

requirements-extras.txt

@@ -0,0 +1,2 @@
+ipython==8.4.0
+celery[redis]==5.1.2

src/rhdzmota/VERSION

@@ -1 +1 @@
-1.2.0
+2.0.0

src/rhdzmota/celery.py

@@ -0,0 +1,12 @@
+from celery import Celery
+
+from .settings import (
+    RHDZMOTA_CELERY_BROKER_HOST
+)
+
+
+# Celery application instance
+app = Celery(
+    "tasks",
+    broker=f"redis://{RHDZMOTA_CELERY_BROKER_HOST}",
+)

src/rhdzmota/celery_worker_hello.py

@@ -0,0 +1,18 @@
+
+from .celery import app
+from .heartbeat.daemon import Daemon
+
+
+@app.task
+def daemon_publisher(**kwargs):
+    print(kwargs)
+
+
+# Daemon instance
+daemon = Daemon(name="hello-world", publisher=daemon_publisher)
+daemon.broadcast()
+
+
+@app.task
+def worker(name: str) -> str:
+    return f"Hello, {name or 'world'}!"

src/rhdzmota/cli.py

@@ -1,3 +1,4 @@
+import time
 from typing import Optional
 
 from .settings import logger_manager
@@ -10,11 +11,20 @@
 
 class CLI(CLIBase):
 
-    @CLIBase.Formatter()
-    def hello(self, world: Optional[str] = None):
-        logger.debug("CLI Hello command execution.")
+    @CLIBase.Formatter(default=str)
+    def hello(self, world: Optional[str] = None, sleep: int = 1, delegate: bool = False):
         world = world or "world"
-        return f"Hello, {world}!"
+        time.sleep(sleep)
+        try:
+            from .celery_worker_hello import worker
+
+            if delegate:
+                return worker.delay(name=world)
+
+            logger.debug("CLI Hello command execution.")
+            return worker(name=world)
+        except ImportError:
+            return f"Hello, {world}!"
 
     @CLIBase.Formatter()
     def gist(

src/rhdzmota/heartbeat/daemon.py

@@ -0,0 +1,73 @@
+from threading import Thread
+from typing import Any, Callable, Optional, Union, TYPE_CHECKING
+
+from .daemon_heart import DaemonHeart
+
+
+if TYPE_CHECKING:
+    # TODO: Quickfix guided by https://stackoverflow.com/questions/63714223/correct-type-annotation-for-a-celery-task
+    from celery.task import Task
+    from celery.local import PromiseProxy
+
+    CELERY_TASK_TYPE = Union[Task, PromiseProxy]
+else:
+    CELERY_TASK_TYPE = Any
+
+
+class Daemon:
+
+    @staticmethod
+    def summon(name: str, publisher: CELERY_TASK_TYPE, **heart_configs):
+        def decorator(worker: CELERY_TASK_TYPE):
+            def wrapper(*args, **kwargs):
+                # Initialize the daemon and broadcast heartbeat
+                daemon = Daemon(name=name, publisher=publisher, **heart_configs)
+                daemon.broadcast()
+                # Execute the worker function
+                output = worker.delay(*args, **kwargs)
+                # Kill the daemon
+                daemon.kill()
+                return output
+            return wrapper
+        return decorator
+
+    def __init__(
+            self,
+            name: str,
+            publisher: CELERY_TASK_TYPE,
+            **heart_configs
+    ):
+        self.name = name
+        self.publisher = publisher
+        self.broadcast_thread: Optional[Thread] = None
+        self.heart = DaemonHeart.beat(
+            app_name=name,
+            **{  # type: ignore
+                "mode": DaemonHeart.Mode.MONITOR,
+                "enable_beat_logs": True,
+                "enable_pulse_monitor": True,
+                "pulse_monitor_frequency": None,
+                "pulse_monitor_sensibility_factor": 1.5,
+                **heart_configs
+            }
+        )
+
+    def kill(self):
+        # Stop heart
+        self.heart.stroke()
+        self.heart.join()
+        self.heart.close()
+        # Stop broadcast
+        self.broadcast_thread.join() if self.broadcast_thread is not None else None
+
+    def broadcast(self):
+        self.broadcast_thread: Thread = Thread(
+            target=lambda: self._broadcast(),
+            daemon=True
+        )
+        self.broadcast_thread.start()
+
+    def _broadcast(self):
+        while not self.heart.no_pulse.is_set():
+            payload = self.heart.queue_beat_logs.get(block=True)
+            self.publisher.delay(**payload)

src/rhdzmota/utils/heartbeat.py → src/rhdzmota/heartbeat/daemon_heart.py

@@ -6,7 +6,7 @@
 from multiprocessing import Event, Queue, Process, Value
 from threading import Thread
 
-from .runtime import system_describe
+from ..utils.runtime import system_describe
 
 
 class DaemonHeart(Process):

src/rhdzmota/settings.py

@@ -123,6 +123,13 @@ def singleton(cls, overwrite_config_dictionary: Optional[Dict] = None) -> 'Logge
 logger_manager = LoggerManager.singleton()
 logger = logger_manager.get_logger(name=__name__)
 
+# Celery
+
+RHDZMOTA_CELERY_BROKER_HOST = get_environ_variable(
+    name="RHDZMOTA_CELERY_BROKER_HOST",
+    default="127.0.0.1"
+)
+
 # JWT
 
 JWT_ENCRYPTION_ALGORITHM: str = get_environ_variable(

worker-hello.sh

@@ -0,0 +1 @@
+celery -A rhdzmota.celery_worker_hello worker --loglevel=INFO --concurrency=2 -O fair -P prefork