---

Task Queue System (FastAPI + PostgreSQL + RabbitMQ) – v0.3

Produktionsreifes Beispielsystem für Job/Task‑Verarbeitung mit Python 3.12, FastAPI, SQLAlchemy (async), PostgreSQL, RabbitMQ (aio‑pika), argparse‑CLI und migrationsfähiger DB‑Struktur via Alembic.

Inhalt

• Features
• Architektur
• Voraussetzungen
• Schnellstart (lokal)
• Konfiguration
• API
• CLI
• Worker
• Alembic (Migrationen)
• Troubleshooting
• Nächste Schritte

---

Features

• JobType → Tasks (z. B. MailCheck):
  MXValidation, SPFValidation, TLSCheck, CertValidity, MTASTSCheck, TLSRPTCheck
• Task‑Abhängigkeiten (DAG):
  TLSCheck & CertValidity warten auf MXValidation; andere Tasks laufen unabhängig.
• Persistenz: Jobs/Tasks in PostgreSQL (UUID, JSONB, Enums, Timestamps).
• RabbitMQ (Topic‑Exchange): Routing per Key <JobType>.<TaskName>.
• Worker (async) mit Retries & Exponential‑Backoff; Status‑Updates in DB.
• API (FastAPI): Job erstellen & Status abfragen.
• CLI (argparse): create-job und get-job.
• Alembic: saubere Baseline & künftige Migrationen (async env.py, Online‑Modus).

---

Architektur

Client/CLI → FastAPI → PostgreSQL
                   ↘ RabbitMQ (Topic)
                      ↘ Worker (MX/SPF/TLS/Cert/MTA-STS/TLS-RPT)

• Job erstellen: API legt Job + Tasks (abhängig vom JobType) an, published Tasks via RabbitMQ.
• Worker: Konsumieren Task‑Messages, führen Checks aus (DNS, SMTP STARTTLS, HTTP), speichern Ergebnisse in DB, releasen abhängige Tasks.
• Status: Job‑Status wird aus Task‑Status aggregiert.

---

Voraussetzungen

• Python 3.12
• PostgreSQL 16 (lokal oder via Docker)
• RabbitMQ 3 (lokal oder via Docker)
• Optional: poetry oder uv (Pakete/venv)

---

Schnellstart (lokal)

1) Services (optional per Docker)

docker-compose up -d
# Ports:
# PostgreSQL → 5432
# RabbitMQ   → 5672 (AMQP), 15672 (UI)

2) Virtuelles Environment & Pakete

python3.12 -m venv .venv
source .venv/bin/activate
pip install -U pip wheel
pip install fastapi uvicorn[standard] sqlalchemy asyncpg pydantic aio-pika typer httpx python-dotenv dnspython alembic

3) Umgebungsvariablen

export DATABASE_URL="postgresql+asyncpg://postgres:postgres@localhost:5432/tasks"
export RABBITMQ_URL="amqp://guest:guest@localhost/"
export RABBITMQ_EXCHANGE="tasks"
export MAX_RETRIES=3
export DNS_TIMEOUT=3.0
export DNS_LIFETIME=5.0
export SMTP_TIMEOUT=10.0
export HTTP_TIMEOUT=8.0

4) Alembic – Baseline anwenden

Wichtig: Die App führt kein create_all() mehr aus; Tabellen werden ausschließlich über Alembic erstellt.

# (Bereits vorhandene DB ohne Tabellen)
PYTHONPATH=. alembic upgrade head

5) API starten

uvicorn app.api:app --host 0.0.0.0 --port 8000 --reload

6) Worker starten (jeweils in separaten Terminals)

python -m app.workers.mail_mx_worker
python -m app.workers.mail_spf_worker
python -m app.workers.mail_tls_worker
python -m app.workers.mail_cert_worker
python -m app.workers.mail_mtasts_worker
python -m app.workers.mail_tlsrpt_worker

7) CLI (argparse)

python -m app.cli create-job MailCheck --domain example.com --api-url http://127.0.0.1:8000
python -m app.cli get-job <JOB_ID> --api-url http://127.0.0.1:8000

---

Konfiguration

• .env (optional): Du kannst obige Variablen in eine .env legen und z. B. via python-dotenv laden.
• RHEL‑freundlich: CLI nutzt argparse statt Typer/Click; keine Abhängigkeit von Click‑Versionen.

---

API

POST /jobs

Body:

{
  "job_type": "MailCheck",
  "domain": "example.com",
  "payload": {}
}

Antwort (verkürzt):

{
  "id": "UUID",
  "job_type": "MailCheck",
  "status": "queued|running|success|failed",
  "domain": "example.com",
  "payload": {},
  "tasks": [
    {"id":"UUID","name":"MXValidation","status":"queued", ...},
    ...
  ]
}

GET /jobs/{job_id}

Gibt Job + Task‑Details (Result/Errors) zurück.

---

CLI

Job anlegen

python -m app.cli create-job MailCheck --domain example.com --api-url http://127.0.0.1:8000

Job abfragen

python -m app.cli get-job <JOB_ID> --api-url http://127.0.0.1:8000

Tipp: Setze TASK_QUEUE_API_URL="http://127.0.0.1:8000" und spare dir --api-url.

---

Worker

• MXValidation: DNS MX‑Lookup (dnspython), sortiert nach Preference.
• SPFValidation: TXT SPF‑Erkennung, Basis‑Parsing (Mechanismen, all‑Qualifier).
• TLSCheck: SMTP EHLO → STARTTLS (Port 25); TLS‑Version/Cipher/Zertifikat; nutzt MX‑Hosts (aus MXValidation oder Fallback DNS).
• CertValidity: Zertifikat notBefore/notAfter; Gültigkeit + Tage bis Ablauf.
• MTASTSCheck: _mta-sts.<domain> TXT + https://mta-sts.<domain>/.well-known/mta-sts.txt.
• TLSRPTCheck: _smtp._tls.<domain> TXT → rua=mailto:….

Retry/Backoff: Bei Fehlern retries++, Status QUEUED, erneutes Publish mit Exponential‑Backoff.

---

Alembic (Migrationen)

Async env.py

• Lädt Base.metadata und importiert Modelle (from app import models) – wichtig, damit alle Tabellen in metadata.tables stehen.
• Führt run_migrations() direkt beim Import aus (kein __main__‑Guard).
• Online‑Modus mit create_async_engine(...) und run_sync(...):

  def do_run_migrations(connection: Connection) -> None:
      context.configure(
          connection=connection,
          target_metadata=Base.metadata,
          compare_type=True,
          compare_server_default=True,
          include_schemas=True,
          process_revision_directives=process_revision_directives, # skip empty
      )
      with context.begin_transaction():
          context.run_migrations()
  await async_conn.run_sync(do_run_migrations)
  

Baseline anwenden

PYTHONPATH=. alembic upgrade head

Neue Revision erzeugen

1. Modelle ändern (z. B. Spalte ergänzen).
2. Revision:

PYTHONPATH=. alembic revision --autogenerate -m "add tasks.priority"

3. Prüfen und anwenden:

PYTHONPATH=. alembic upgrade head

Hinweis (Enums): Beim Erweitern von Enum‑Werten ggf. manuell ergänzen:

op.execute("ALTER TYPE jobstatus ADD VALUE IF NOT EXISTS 'paused'")

---

Troubleshooting

Alembic erzeugt leere Revisionen

• Prüfe, ob env.py beim Import run_migrations() aufruft (kein if __name__ == "__main__").
• Sicherstellen: context.configure(connection=..., target_metadata=...) wird im Online‑Pfad gesetzt.
• Modelle importieren (from app import models), damit Base.metadata.tables nicht leer ist.
• Kein Offline‑Run (--sql vermeiden); Offline‑Pfad ggf. bewusst deaktivieren.
• Starte Alembic aus dem Projekt‑Root mit PYTHONPATH=., damit from app ... funktioniert.

MissingGreenlet beim API‑Zugriff

• Ursache: Lazy‑Loading einer Beziehung in async Kontext (z. B. job.tasks).
• Fix: Beziehungen standardmäßig eager laden (lazy="selectin") oder im Query options(selectinload(Job.tasks)) setzen.

Typer/Click‑Fehler in CLI (RHEL)

• CLI nutzt argparse (keine Abhängigkeit zu Typer/Click).

---

Nächste Schritte

• Makefile/Scripts ergänzen (z. B. make migrate, make revision m="...").
• Docker Entrypoint: vor App‑Start alembic upgrade head.
• Observability: Logging strukturieren; Metriken/Tracing (optional).
• Tests: Unit‑/Integration‑Tests (z. B. Worker‑Probes mit Mocks, DNS/SMTP/HTTP Timeouts).

---