implement handler exec: publish, childs, pause - unpause

Author: leodisarli

README.md

@@ -0,0 +1,45 @@
+# importing the lib
+from omniq.client import OmniqClient
+
+# creating your handler (ctx will have all the job information and actions)
+def document_worker(ctx):
+    
+    # getting the data from payload
+    document_id = ctx.payload["document_id"]
+    pages = ctx.payload["pages"]
+
+    # creating the unique key to track the childs
+    completion_key = f"document:{document_id}"
+
+    print(f"[document_worker] Initializing completion for {pages} pages")
+
+    # calling the childs initialization
+    ctx.exec.childs_init(completion_key, pages)
+
+    # publishing 5 jobs on the pages queue
+    for page in range(1, pages + 1):
+        # ctx.exec also have the publisher ready to use
+        ctx.exec.publish(
+            queue="pages",
+            payload={
+                "document_id": document_id,
+                "page": page,
+                "completion_key": completion_key,
+            },
+        )
+
+    print("[document_worker] All page jobs published.")
+
+# creating OmniQ passing redis information
+omniq = OmniqClient(
+    host="omniq-redis",
+    port=6379,
+)
+
+# creating the consumer that will listen and execute the actions in your handler
+omniq.consume(
+    queue="documents",
+    handler=document_worker,
+    verbose=True,
+    drain=False,
+)

examples/childs/consumer1.py

@@ -0,0 +1,40 @@
+import time
+
+# importing the lib
+from omniq.client import OmniqClient
+
+# creating your handler (ctx will have all the job information and actions)
+def page_worker(ctx):
+
+    page = ctx.payload["page"]
+    # getting the unique key to track the childs
+    completion_key = ctx.payload["completion_key"]
+
+    print(f"[page_worker] Processing page {page} (job_id={ctx.job_id})")
+    time.sleep(1.5)
+
+    # acking itself as a child the number of remaining jobs are returned so we can say when the last job was executed
+    remaining = ctx.exec.child_ack(completion_key)
+
+    print(f"[page_worker] Page {page} done. Remaining={remaining}")
+    
+
+    # remaining will be 0 ONLY when this is the last job
+    # will return > 0 when are still jobs to process
+    # and -1 if something goes wrong with the counter
+    if remaining == 0:
+        print("[page_worker] Last page finished.")
+
+# creating OmniQ passing redis information
+omniq = OmniqClient(
+    host="omniq-redis",
+    port=6379,
+)
+
+# creating the consumer that will listen and execute the actions in your handler
+omniq.consume(
+    queue="pages",
+    handler=page_worker,
+    verbose=True,
+    drain=False,
+)

examples/childs/consumer2.py

@@ -0,0 +1,18 @@
+# importing the lib
+from omniq.client import OmniqClient
+
+# creating OmniQ passing redis information
+omniq = OmniqClient(
+    host="omniq-redis",
+    port=6379,
+)
+
+# publishing the job
+job_id = omniq.publish(
+    queue="documents",
+    payload={
+        "document_id": "doc-123", # this will be our unique key to initiate childs and tracking then until completion
+        "pages": 5, # each page must be completed before something happen
+    },
+)
+print("OK", job_id)

examples/childs/publish.py

@@ -0,0 +1,60 @@
+import time
+
+# importing the lib
+from omniq.client import OmniqClient
+
+# creating your handler (ctx will have all the job information and actions)
+def pause_unpause_example(ctx):
+    print("Waiting 2 seconds")
+
+    # checking if this queue it is paused (spoiler: it's not)
+    is_paused = ctx.exec.is_paused(
+        "test"
+    )
+    print("Is paused", is_paused)
+    time.sleep(2)
+
+
+    print("Pausing")
+
+    # pausing this queue (this job it's and others active jobs will be not affected but not new job will be start until queue is resumed)
+    ctx.exec.pause(
+        "test"
+    )
+
+    # checking again now is suposed to be paused
+    is_paused = ctx.exec.is_paused(
+        "test"
+    )
+    print("Is paused", is_paused)
+    time.sleep(2)
+
+    print("Resuming")
+
+    # resuming this queue (all other workers can process jobs again)
+    ctx.exec.resume(
+        "test"
+    )
+
+    # checking again and is suposed to be resumed
+    is_paused = ctx.exec.is_paused(
+        "test"
+    )
+    print("Is paused", is_paused)
+    time.sleep(2)
+
+    print("Done")
+
+# creating OmniQ passing redis information
+omniq = OmniqClient(
+    host="omniq-redis",
+    port=6379,
+)
+
+# creating the consumer that will listen and execute the actions in your handler
+omniq.consume(
+    queue="test",
+    handler=pause_unpause_example,
+    verbose=True,
+    drain=False,
+)

examples/consumer.py

@@ -1,11 +1,15 @@
+# importing the lib
 from omniq.client import OmniqClient
 
+# creating OmniQ passing redis information
 uq = OmniqClient(
     host="omniq-redis",
     port=6379,
 )
+
+# publishing the job
 job_id = uq.publish(
-    queue="demo",
+    queue="test",
     payload={"hello": "world"},
     timeout_ms=30_000
 )

examples/pause_resume/consumer-pause.py

@@ -0,0 +1,24 @@
+import time
+
+# importing the lib
+from omniq.client import OmniqClient
+
+# creating your handler (ctx will have all the job information and actions)
+def my_actions(ctx):
+    print("Waiting 2 seconds")
+    time.sleep(2)
+    print("Done")
+
+# creating OmniQ passing redis information
+omniq = OmniqClient(
+    host="omniq-redis",
+    port=6379,
+)
+
+# creating the consumer that will listen and execute the actions in your handler
+omniq.consume(
+    queue="demo",
+    handler=my_actions,
+    verbose=True,
+    drain=False,
+)

examples/publish.py examples/pause_resume/publish.pyexamples/publish.py renamed to examples/pause_resume/publish.py

@@ -0,0 +1,17 @@
+# importing the lib
+from omniq.client import OmniqClient
+
+# creating OmniQ passing redis information
+omniq = OmniqClient(
+    host="omniq-redis",
+    port=6379,
+)
+
+# publishing the job
+job_id = omniq.publish(
+    queue="demo",
+    payload={"hello": "world"},
+    timeout_ms=30_000
+)
+
+print("OK", job_id)

examples/simple/consumer.py

@@ -1,6 +1,6 @@
 [project]
 name = "omniq"
-version = "1.4.1"
+version = "1.5.0"
 authors = [
   { name="Not Empty Foundation", email="dev@not-empty.org" },
 ]