280 lines
7.7 KiB
Markdown
280 lines
7.7 KiB
Markdown
***
|
||
|
||
# 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).
|
||
|
||
***
|