prod msg enq
This commit is contained in:
@@ -89,18 +89,18 @@ Queue args for `orders.q`:
|
|||||||
|
|
||||||
## Step 3: Producer (20 min)
|
## Step 3: Producer (20 min)
|
||||||
|
|
||||||
- [ ] `producer.py` should publish 5 JSON orders to routing key `order.created`.
|
- [x] `producer.py` should publish 5 JSON orders to routing key `order.created`.
|
||||||
- [ ] Each message should include:
|
- [x] Each message should include:
|
||||||
- [ ] `order_id`
|
- [x] `order_id`
|
||||||
- [ ] `user_id`
|
- [x] `user_id`
|
||||||
- [ ] `amount`
|
- [x] `amount`
|
||||||
- [ ] `should_fail` (set true for at least 1 order)
|
- [x] `should_fail` (set true for at least 1 order)
|
||||||
- [ ] `retry_count` (start at 0)
|
- [x] `retry_count` (start at 0)
|
||||||
- [ ] Mark messages persistent.
|
- [x] Mark messages persistent.
|
||||||
|
|
||||||
Success check:
|
Success check:
|
||||||
|
|
||||||
- [ ] Running `uv run producer.py` prints “Published order …” 5 times.
|
- [x] Running `uv run producer.py` prints “Published order …” 5 times.
|
||||||
|
|
||||||
## Step 4: Worker (35 min)
|
## Step 4: Worker (35 min)
|
||||||
|
|
||||||
|
|||||||
@@ -9,6 +9,13 @@ channel = connection.channel()
|
|||||||
|
|
||||||
# this is A DECLARE, basically so can run multuple times
|
# this is A DECLARE, basically so can run multuple times
|
||||||
# .x = exchange
|
# .x = exchange
|
||||||
|
|
||||||
|
# what does durable mean?
|
||||||
|
# when broke restarts
|
||||||
|
# even if I don't run this IaC like code to create the q
|
||||||
|
# the q will exist
|
||||||
|
# it basically commits the q defns to disk
|
||||||
|
# not the message
|
||||||
channel.exchange_declare(
|
channel.exchange_declare(
|
||||||
exchange="orders.x", exchange_type="direct", durable=True)
|
exchange="orders.x", exchange_type="direct", durable=True)
|
||||||
# .dlx = dead letter exchange for failed messages
|
# .dlx = dead letter exchange for failed messages
|
||||||
|
|||||||
+38
@@ -0,0 +1,38 @@
|
|||||||
|
import pika
|
||||||
|
import json
|
||||||
|
|
||||||
|
conn = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
|
||||||
|
ch = conn.channel()
|
||||||
|
|
||||||
|
# can redeclare qs, but since they are in main no need to
|
||||||
|
# the op would've been idempotent if I did though
|
||||||
|
|
||||||
|
# random objets to be conv to json
|
||||||
|
orders = [
|
||||||
|
{"order_id": 1001, "user_id": "u1", "amount": 49.99,
|
||||||
|
"should_fail": False, "retry_count": 0},
|
||||||
|
{"order_id": 1002, "user_id": "u2", "amount": 19.50,
|
||||||
|
"should_fail": False, "retry_count": 0},
|
||||||
|
{"order_id": 1003, "user_id": "u3", "amount": 99.00,
|
||||||
|
"should_fail": True, "retry_count": 0},
|
||||||
|
{"order_id": 1004, "user_id": "u4", "amount": 10.00,
|
||||||
|
"should_fail": False, "retry_count": 0},
|
||||||
|
{"order_id": 1005, "user_id": "u5", "amount": 75.25,
|
||||||
|
"should_fail": False, "retry_count": 0},
|
||||||
|
]
|
||||||
|
|
||||||
|
for order in orders:
|
||||||
|
body = json.dumps(order)
|
||||||
|
ch.basic_publish(
|
||||||
|
exchange="orders.x",
|
||||||
|
routing_key="order.created", # as in the bindings
|
||||||
|
body=body,
|
||||||
|
properties=pika.BasicProperties(
|
||||||
|
delivery_mode=2, # persistent msg, committed to disk before ack
|
||||||
|
content_type="application/json"
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
print(f"published order {order['order_id']}")
|
||||||
|
|
||||||
|
conn.close()
|
||||||
|
|||||||
Reference in New Issue
Block a user