-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathpydantic_demo.py
More file actions
60 lines (45 loc) · 1.74 KB
/
pydantic_demo.py
File metadata and controls
60 lines (45 loc) · 1.74 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from pydantic import BaseModel, Field
from modelq import ModelQ
from redis import Redis
class AddIn(BaseModel):
a: int = Field(ge=0)
b: int = Field(ge=0)
class AddOut(BaseModel):
total: int
redis_client = Redis(host="localhost", port=6379, db=0)
mq = ModelQ(redis_client = redis_client)
@mq.task(schema=AddIn, returns=AddOut)
def add(payload: AddIn) -> AddOut:
print(f"Processing addition: {payload.a} + {payload.b}")
time.sleep(10) # Simulate some processing time
return AddOut(total=payload.a + payload.b)
@mq.task()
def sub(a: int, b: int):
print(f"Processing subtraction: {a} - {b}")
return a - b
@mq.task()
def image_task(params: dict):
print(f"Processing image task with params: {params}")
# Simulate image processing
return "Image processed successfully"
job = add(a=3, b=4) # ✨ validated on the spot
job2 = sub(a=10, b=5) # ✨ no schema validation, just a simple task
task = image_task({"image": "example.png"}) # ✨ no schema validation, just a simple task
task2 = image_task(params={"image": "example.png"})
import time
if __name__ == "__main__":
mq.start_workers()
# Keep the worker running indefinitely
try:
while True:
output = job.get_result(mq.redis_client,returns=AddOut)
print(f"Result of addition: {output}")
print(type(output))
print(f"Result of addition (total): {output.total}")
output2 = job2.get_result(mq.redis_client)
print(f"Result of subtraction: {output2}")
output3 = task.get_result(mq.redis_client)
print(f"Result of image task: {output3}")
time.sleep(1)
except KeyboardInterrupt:
print("\nGracefully shutting down...")