app.py

"""
Background Job Queue — Flask + threading-based worker (same pattern as Celery, no broker dependency)
Simulates: user submits a file-processing job -> API returns immediately -> worker processes async -> status is pollable.
Run: python app.py
"""
from flask import Flask, request, jsonify
import threading, queue, uuid, time, sqlite3, os, datetime

app = Flask(__name__)
DB_PATH = os.path.join(os.path.dirname(__file__), "jobs.db")
JOB_QUEUE = queue.Queue()


def get_db():
    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    conn = get_db()
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS jobs (
            id TEXT PRIMARY KEY,
            job_type TEXT NOT NULL,
            payload TEXT,
            status TEXT NOT NULL DEFAULT 'queued',
            result TEXT,
            created_at TEXT NOT NULL,
            finished_at TEXT
        );
        """
    )
    conn.commit()
    conn.close()


def process_job(job_id, job_type, payload):
    """Simulates real work — e.g. resizing an image, parsing a file, sending a batch email."""
    time.sleep(2)  # stand-in for real processing time
    if job_type == "word_count":
        result = f"word_count={len(str(payload).split())}"
    elif job_type == "uppercase":
        result = str(payload).upper()
    else:
        result = "unknown job_type, no-op"
    return result


def worker_loop():
    while True:
        job_id, job_type, payload = JOB_QUEUE.get()
        db = get_db()
        db.execute("UPDATE jobs SET status = ? WHERE id = ?", ("processing", job_id))
        db.commit()
        try:
            result = process_job(job_id, job_type, payload)
            db.execute(
                "UPDATE jobs SET status = ?, result = ?, finished_at = ? WHERE id = ?",
                ("completed", result, datetime.datetime.utcnow().isoformat(), job_id),
            )
        except Exception as e:
            db.execute(
                "UPDATE jobs SET status = ?, result = ? WHERE id = ?",
                ("failed", str(e), job_id),
            )
        db.commit()
        db.close()
        JOB_QUEUE.task_done()


@app.route("/api/jobs", methods=["POST"])
def submit_job():
    data = request.get_json(force=True)
    job_type = data.get("job_type")
    payload = data.get("payload", "")
    if not job_type:
        return jsonify({"error": "job_type is required"}), 400

    job_id = str(uuid.uuid4())
    db = get_db()
    db.execute(
        "INSERT INTO jobs (id, job_type, payload, status, created_at) VALUES (?, ?, ?, ?, ?)",
        (job_id, job_type, str(payload), "queued", datetime.datetime.utcnow().isoformat()),
    )
    db.commit()
    db.close()

    JOB_QUEUE.put((job_id, job_type, payload))
    return jsonify({"job_id": job_id, "status": "queued"}), 202


@app.route("/api/jobs/<job_id>", methods=["GET"])
def job_status(job_id):
    db = get_db()
    row = db.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone()
    db.close()
    if not row:
        return jsonify({"error": "not found"}), 404
    return jsonify(dict(row))


if __name__ == "__main__":
    init_db()
    worker_thread = threading.Thread(target=worker_loop, daemon=True)
    worker_thread.start()
    app.run(debug=False, port=5004)  # debug=False so the reloader doesn't spawn duplicate workers