83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
|
||
# app/cli.py
|
||
import argparse
|
||
import sys
|
||
import json
|
||
import os
|
||
import httpx
|
||
|
||
DEFAULT_API_URL = os.environ.get("TASK_QUEUE_API_URL", "http://127.0.0.1:8000")
|
||
|
||
def cmd_create_job(args: argparse.Namespace) -> int:
|
||
"""
|
||
Legt einen neuen Job an (MailCheck) und gibt Status/Tasks aus.
|
||
"""
|
||
payload = {} # ggf. später erweitern
|
||
body = {"job_type": args.job_type, "domain": args.domain, "payload": payload}
|
||
try:
|
||
with httpx.Client(base_url=args.api_url, timeout=15) as client:
|
||
resp = client.post("/jobs", json=body)
|
||
resp.raise_for_status()
|
||
job = resp.json()
|
||
except httpx.HTTPError as e:
|
||
print(f"[ERROR] API-Request fehlgeschlagen: {e}", file=sys.stderr)
|
||
return 1
|
||
|
||
print(f"Job {job['id']} ({job['job_type']}) Domain: {job['domain']} Status: {job['status']}")
|
||
for t in job.get("tasks", []):
|
||
print(f"- Task {t['name']}: {t['status']} (id={t['id']})")
|
||
return 0
|
||
|
||
def cmd_get_job(args: argparse.Namespace) -> int:
|
||
"""
|
||
Holt einen bestehenden Job und zeigt Status/Tasks.
|
||
"""
|
||
try:
|
||
with httpx.Client(base_url=args.api_url, timeout=15) as client:
|
||
resp = client.get(f"/jobs/{args.job_id}")
|
||
resp.raise_for_status()
|
||
job = resp.json()
|
||
except httpx.HTTPError as e:
|
||
print(f"[ERROR] API-Request fehlgeschlagen: {e}", file=sys.stderr)
|
||
return 1
|
||
|
||
print(json.dumps(job, indent=2))
|
||
return 0
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(
|
||
prog="tq",
|
||
description="Task-Queue CLI (FastAPI + PostgreSQL + RabbitMQ) – RHEL-freundlich mit argparse",
|
||
)
|
||
parser.add_argument(
|
||
"--api-url", default=DEFAULT_API_URL,
|
||
help=f"Basis-URL der API (Default: {DEFAULT_API_URL})"
|
||
)
|
||
|
||
sub = parser.add_subparsers(dest="command", required=True)
|
||
|
||
# create-job
|
||
p_create = sub.add_parser("create-job", help="Neuen Job anlegen")
|
||
p_create.add_argument("job_type", help="JobType, z. B. MailCheck")
|
||
p_create.add_argument("--domain", "-d", required=True, help="Domain für den Job (z. B. example.com)")
|
||
p_create.set_defaults(func=cmd_create_job)
|
||
|
||
# get-job
|
||
p_get = sub.add_parser("get-job", help="Job-Status abrufen")
|
||
p_get.add_argument("job_id", help="Job-ID (UUID)")
|
||
p_get.set_defaults(func=cmd_get_job)
|
||
|
||
return parser
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = build_parser()
|
||
args = parser.parse_args(argv)
|
||
# Subcommand dispatch
|
||
if hasattr(args, "func"):
|
||
return args.func(args)
|
||
parser.print_help()
|
||
return 2
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|
||
|