server/worker.py - laba4 - mmr97/laba1_API

server/worker.py
68 строк2 KB
mmr97
laba4
01 дек 2025, 16:56
01 дек 2025, 16:56b975a9d
import pika
import json
import uuid
import os

STORE_FILE = "store.json"
if not os.path.exists(STORE_FILE):
    with open(STORE_FILE, "w") as f:
        json.dump({}, f)

RABBITMQ_URL = "amqp://test:test@localhost:5672/"

def save_processed(id, data):
    with open(STORE_FILE, "r") as f:
        db = json.load(f)
    db[id] = data
    with open(STORE_FILE, "w") as f:
        json.dump(db, f, indent=2)

def get_processed(id):
    with open(STORE_FILE, "r") as f:
        db = json.load(f)
    return db.get(id, None)

def callback(ch, method, properties, body):
    request = json.loads(body)
    id = request.get("id")
    action = request.get("action")
    data = request.get("data")
    auth = request.get("auth")

    # идемпотентность
    cached = get_processed(id)
    if cached:
        print(f"♻ Returning cached response for {id}")
        ch.basic_ack(delivery_tag=method.delivery_tag)
        return

    # простая обработка действий
    if action == "create_task":
        response_data = {
            "task_id": uuid.uuid4().hex[:8],
            "title": data.get("title"),
            "description": data.get("description", ""),
            "status": "todo"
        }
    else:
        response_data = {"message": "Action not supported"}

    response = {
        "correlation_id": id,
        "status": "ok",
        "data": response_data,
        "error": None
    }

    save_processed(id, response)
    print(f"Processed {action} with correlation_id={id}")

    ch.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(pika.URLParameters(RABBITMQ_URL))
channel = connection.channel()
channel.queue_declare(queue="api.requests", durable=True)

print("Worker started. Waiting for messages...")
channel.basic_consume(queue="api.requests", on_message_callback=callback)
channel.start_consuming()