Add task experimentation

compose.yml

@@ -1,5 +1,7 @@
 services:
   app:
+    depends_on:
+      - rabbitmq
     build: 
       context: .
       target: development
@@ -15,3 +17,11 @@ services:
       - .:/app
     tty: true
     stdin_open: true
+  rabbitmq:
+    image: 'rabbitmq:4.0-management'
+    environment:
+      - 'RABBITMQ_DEFAULT_USER=admin'
+      - 'RABBITMQ_DEFAULT_PASS=secret'
+    ports:
+      - '5672:5672'
+      - '15672:15672'

env.example

@@ -1,4 +1,6 @@
 UID=YOUR_UID
 GID=YOUR_GID
-POETRY_VERSION=1.5.1
+POETRY_VERSION=1.8.4
 DEV=true
+
+RABBITMQ_URL="amqp://admin:secret@rabbitmq:5672"

pyproject.toml

@@ -20,6 +20,7 @@ typer = "^0.12.5"
 uuid6 = "^2024.7.10"
 python-ulid = "^2.7.0"
 jinja2 = "^3.1.4"
+dramatiq = {extras = ["rabbitmq", "watch"], version = "^1.17.0"}
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.0.2"

requirements.txt

@@ -1 +1 @@
-poetry==1.5.1
+poetry==1.8.4

tasks/README.md

@@ -0,0 +1,25 @@
+## Task experimentation with dramatiq
+
+Three terminals...
+
+1. RabbitMQ
+
+```sh
+docker compose up rabbitmq
+```
+
+1. Dramatiq workers
+
+```sh
+docker compose run app bash
+poetry run dramatiq tasks.test_task
+```
+
+1. Send messages
+
+```sh
+docker compose exec app bash
+poetry run python
+>>> from tasks.test_task import sleep_and_say_hi
+>>> [sleep_and_say_hi.send() for i in range(10)]
+```

tasks/broker.py

@@ -0,0 +1,23 @@
+import os
+from typing import Self
+
+import dramatiq
+from dramatiq.brokers.rabbitmq import RabbitmqBroker
+
+class DramatiqRabbitmqBroker:
+
+    @classmethod
+    def from_env(cls) -> Self:
+        rabbitmq_url = os.environ.get("RABBITMQ_URL")
+        if not rabbitmq_url:
+            raise ValueError("RABBITMQ_URL environment variable not configured!")
+        return cls(rabbitmq_url)
+
+    def __init__(self, rabbitmq_url: str):
+        self.rabbitmq_broker = RabbitmqBroker(url=rabbitmq_url)
+        dramatiq.set_broker(self.rabbitmq_broker)
+
+    def set_up_queue(self, name: str):
+        if name not in self.rabbitmq_broker.get_declared_queues():
+            self.rabbitmq_broker.declare_queue(name)
+

tasks/test_task.py

@@ -0,0 +1,17 @@
+import time
+from random import randrange
+
+import dramatiq
+
+from tasks.broker import DramatiqRabbitmqBroker
+
+QUEUE_NAME = "sleep"
+
+BROKER = DramatiqRabbitmqBroker.from_env()
+BROKER.set_up_queue(QUEUE_NAME)
+
+@dramatiq.actor(queue_name=QUEUE_NAME)
+def sleep_and_say_hi() -> None:
+    sleep_period = randrange(1, 10)
+    time.sleep(sleep_period)
+    print(f"Hi, I slept {sleep_period} seconds!")