***

# 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)

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

### 2) Virtuelles Environment & Pakete

```bash
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

```bash
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.

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

### 5) API starten

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

### 6) Worker starten (jeweils in separaten Terminals)

```bash
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)

```bash
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:**

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

**Antwort (verkürzt):**

```json
{
  "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

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

### Job abfragen

```bash
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(...)`:
    ```python
    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

```bash
PYTHONPATH=. alembic upgrade head
```

### Neue Revision erzeugen

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

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

3.  Prüfen und anwenden:

```bash
PYTHONPATH=. alembic upgrade head
```

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

```python
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).

***