producer.py

import json
import logging
import os

from dotenv import load_dotenv
from flask import Flask, jsonify, request
import redis

load_dotenv()

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
PORT_PRODUCER = int(os.getenv("PORT_PRODUCER", "8080"))
QUEUE_NAME = "task_queue"

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("producer")

rdb = redis.Redis.from_url(REDIS_URL, decode_responses=True)

app = Flask(__name__)


@app.route("/enqueue", methods=["POST"])
def enqueue():
    task = request.get_json(silent=True)
    if not isinstance(task, dict):
        return jsonify(error="Bad Request: body must be a JSON object"), 400

    task_type = task.get("type")
    if not task_type:
        return jsonify(error="Bad Request: 'type' is required"), 400

    payload = task.get("payload") or {}
    if not isinstance(payload, dict):
        return jsonify(error="Bad Request: 'payload' must be an object"), 400

    if task_type == "send_email":
        if not payload.get("to") or not payload.get("subject"):
            return jsonify(error="Bad Request: send_email requires payload.to and payload.subject"), 400

    normalized = {
        "type": task_type,
        "payload": payload,
        "retries": int(task.get("retries", 0)),
    }

    try:
        queue_len = rdb.rpush(QUEUE_NAME, json.dumps(normalized))
    except redis.RedisError as exc:
        log.exception("Redis push failed")
        return jsonify(error=f"Internal server error: {exc}"), 500

    log.info("Enqueued %s task. Queue length: %s", task_type, queue_len)
    return jsonify(message=f"Task of type '{task_type}' has been successfully added to the queue",
                   queue_length=queue_len), 200


if __name__ == "__main__":
    log.info("Starting producer on port %s", PORT_PRODUCER)
    app.run(host="0.0.0.0", port=PORT_PRODUCER)