311 lines
11 KiB
Python
311 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Gemeinsame Funktionen für die Wegwichtel-API-Werkzeuge."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import getpass
|
|
import json
|
|
import os
|
|
import sys
|
|
import traceback
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urljoin
|
|
from urllib.request import Request, urlopen
|
|
|
|
DEFAULT_BASE_URL = "https://wegwichtel.de"
|
|
API_USER_ENV = "API_USER"
|
|
API_PASSWORD_ENV = "API_PASSWORD"
|
|
APPLICATION_JSON = "application/json"
|
|
MAX_AUTH_ATTEMPTS = 3
|
|
REQUEST_TIMEOUT_SECONDS = 120
|
|
|
|
|
|
class ApiError(RuntimeError):
|
|
"""Fehlerantwort der REST-API einschließlich HTTP-Status und Nutzdaten."""
|
|
|
|
def __init__(self, status: int, message: str, payload: Any = None):
|
|
super().__init__(f"HTTP {status}: {message}")
|
|
self.status = status
|
|
self.payload = payload
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Credentials:
|
|
"""Zugangsdaten für eine nach HTTP 401 angeforderte Basic-Authentifizierung."""
|
|
|
|
username: str
|
|
password: str
|
|
|
|
|
|
def credentials_from_environment() -> Credentials | None:
|
|
"""Liest CI-Credentials, ohne den ersten Request vorab zu authentifizieren."""
|
|
username = os.getenv(API_USER_ENV, "").strip()
|
|
password = os.getenv(API_PASSWORD_ENV, "").strip()
|
|
if username and password:
|
|
return Credentials(username, password)
|
|
return None
|
|
|
|
|
|
def _json_bytes(value: dict[str, Any]) -> bytes:
|
|
return json.dumps(value, ensure_ascii=False).encode("utf-8")
|
|
|
|
|
|
def _request_payload(
|
|
json_body: dict[str, Any] | None,
|
|
fields: dict[str, Any] | None,
|
|
files: dict[str, Path] | None,
|
|
) -> tuple[bytes | None, dict[str, str]]:
|
|
headers = {"Accept": APPLICATION_JSON}
|
|
if files:
|
|
data, content_type = encode_multipart(fields or {}, files)
|
|
headers["Content-Type"] = content_type
|
|
return data, headers
|
|
if json_body is not None:
|
|
headers["Content-Type"] = APPLICATION_JSON
|
|
return _json_bytes(json_body), headers
|
|
if fields:
|
|
headers["Content-Type"] = APPLICATION_JSON
|
|
return _json_bytes(fields), headers
|
|
return None, headers
|
|
|
|
|
|
class ApiClient:
|
|
"""Kleine HTTP-Schicht mit dynamischer Basic-Auth-Nachfrage bei HTTP 401."""
|
|
|
|
def __init__(self, base_url: str, username: str | None = None, password: str | None = None):
|
|
self.base_url = base_url.rstrip("/") + "/"
|
|
self.credentials = Credentials(username.strip(), password.strip()) if username and password else None
|
|
|
|
def prompt_credentials(self) -> None:
|
|
environment_credentials = credentials_from_environment()
|
|
if environment_credentials:
|
|
self.credentials = environment_credentials
|
|
return
|
|
|
|
default_user = self.credentials.username if self.credentials else os.getenv(API_USER_ENV, "").strip()
|
|
suffix = f" [{default_user}]" if default_user else ""
|
|
username = input(f"Benutzername{suffix}: ").strip() or default_user
|
|
if not username:
|
|
raise ApiError(401, "Für den geschützten API-Zugriff ist ein Benutzername erforderlich.")
|
|
|
|
password = getpass.getpass("Passwort: ").strip()
|
|
if not password:
|
|
raise ApiError(401, "Für den geschützten API-Zugriff ist ein Passwort erforderlich.")
|
|
self.credentials = Credentials(username, password)
|
|
|
|
def _authorization_header(self) -> str | None:
|
|
if not self.credentials:
|
|
return None
|
|
value = f"{self.credentials.username}:{self.credentials.password}".encode("utf-8")
|
|
token = base64.b64encode(value).decode("ascii")
|
|
return f"Basic {token}"
|
|
|
|
def _request_headers(self, headers: dict[str, str]) -> dict[str, str]:
|
|
request_headers = dict(headers)
|
|
authorization = self._authorization_header()
|
|
if authorization:
|
|
request_headers["Authorization"] = authorization
|
|
return request_headers
|
|
|
|
def _perform_request(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
data: bytes | None,
|
|
headers: dict[str, str],
|
|
expected: tuple[int, ...],
|
|
) -> Any:
|
|
request = Request(
|
|
url,
|
|
data=data,
|
|
headers=self._request_headers(headers),
|
|
method=method.upper(),
|
|
)
|
|
with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
|
|
raw = response.read()
|
|
if response.status not in expected:
|
|
raise ApiError(response.status, decode_error(raw), parse_json(raw))
|
|
return parse_response(raw, response.headers.get("Content-Type", ""))
|
|
|
|
def _handle_http_error(self, error: HTTPError, attempt: int) -> bool:
|
|
if error.code != 401 or attempt >= MAX_AUTH_ATTEMPTS:
|
|
return False
|
|
error.read()
|
|
error.close()
|
|
print("Die API verlangt Zugangsdaten (HTTP 401).", file=sys.stderr)
|
|
self.prompt_credentials()
|
|
return True
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
json_body: dict[str, Any] | None = None,
|
|
fields: dict[str, Any] | None = None,
|
|
files: dict[str, Path] | None = None,
|
|
expected: tuple[int, ...] = (200, 201),
|
|
) -> Any:
|
|
data, headers = _request_payload(json_body, fields, files)
|
|
url = urljoin(self.base_url, path.lstrip("/"))
|
|
|
|
attempt = 0
|
|
while True:
|
|
try:
|
|
return self._perform_request(method, url, data, headers, expected)
|
|
except HTTPError as error:
|
|
if self._handle_http_error(error, attempt):
|
|
attempt += 1
|
|
continue
|
|
raw = error.read()
|
|
raise ApiError(error.code, decode_error(raw), parse_json(raw)) from error
|
|
except URLError as error:
|
|
raise RuntimeError(f"API nicht erreichbar: {error.reason}") from error
|
|
|
|
|
|
def encode_multipart(fields: dict[str, Any], files: dict[str, Path]) -> tuple[bytes, str]:
|
|
"""Erzeugt multipart/form-data; Dateitypen werden absichtlich nicht mitgesendet."""
|
|
boundary = f"----wegwichtel-{uuid.uuid4().hex}"
|
|
chunks: list[bytes] = []
|
|
|
|
for name, value in fields.items():
|
|
if value is None:
|
|
continue
|
|
chunks.extend([
|
|
f"--{boundary}\r\n".encode(),
|
|
f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(),
|
|
str(value).encode("utf-8"),
|
|
b"\r\n",
|
|
])
|
|
|
|
for name, file_path in files.items():
|
|
path = Path(file_path).expanduser().resolve()
|
|
if not path.is_file():
|
|
raise FileNotFoundError(f"Datei nicht gefunden: {path}")
|
|
safe_name = path.name.replace('"', "_")
|
|
chunks.extend([
|
|
f"--{boundary}\r\n".encode(),
|
|
f'Content-Disposition: form-data; name="{name}"; filename="{safe_name}"\r\n\r\n'.encode(),
|
|
path.read_bytes(),
|
|
b"\r\n",
|
|
])
|
|
|
|
chunks.append(f"--{boundary}--\r\n".encode())
|
|
return b"".join(chunks), f"multipart/form-data; boundary={boundary}"
|
|
|
|
|
|
def parse_json(raw: bytes) -> Any:
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return json.loads(raw.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
return None
|
|
|
|
|
|
def parse_response(raw: bytes, content_type: str) -> Any:
|
|
if not raw:
|
|
return None
|
|
if "json" in content_type.lower():
|
|
parsed = parse_json(raw)
|
|
if parsed is not None:
|
|
return parsed
|
|
return raw.decode("utf-8", errors="replace")
|
|
|
|
|
|
def decode_error(raw: bytes) -> str:
|
|
parsed = parse_json(raw)
|
|
if isinstance(parsed, dict):
|
|
return str(parsed.get("message") or parsed.get("error") or parsed)
|
|
text = raw.decode("utf-8", errors="replace").strip()
|
|
return text or "Unbekannter API-Fehler"
|
|
|
|
|
|
def add_common_arguments(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument("--base-url", help="API-Basisadresse, Standard: WEGWICHTEL_URL oder https://wegwichtel.de")
|
|
parser.add_argument("--username", help="Benutzername; Passwort wird bei HTTP 401 sicher abgefragt")
|
|
|
|
|
|
def client_from_args(args: argparse.Namespace) -> ApiClient:
|
|
base_url = args.base_url or os.getenv("WEGWICHTEL_URL") or DEFAULT_BASE_URL
|
|
return ApiClient(base_url, args.username)
|
|
|
|
|
|
def ask(label: str, value: Any = None, *, required: bool = False, cast: Callable[[str], Any] = str) -> Any:
|
|
if value is not None:
|
|
return cast(str(value))
|
|
while True:
|
|
raw = input(f"{label}: ").strip()
|
|
if raw:
|
|
try:
|
|
return cast(raw)
|
|
except (TypeError, ValueError) as error:
|
|
print(f"Ungültiger Wert: {error}", file=sys.stderr)
|
|
continue
|
|
if not required:
|
|
return None
|
|
print("Dieses Feld ist erforderlich.", file=sys.stderr)
|
|
|
|
|
|
def ask_with_default(label: str, current: Any, cast: Callable[[str], Any] = str) -> Any:
|
|
shown = "" if current is None else str(current)
|
|
while True:
|
|
raw = input(f"{label} [{shown}]: ").strip()
|
|
if not raw:
|
|
return current
|
|
try:
|
|
return cast(raw)
|
|
except (TypeError, ValueError) as error:
|
|
print(f"Ungültiger Wert: {error}", file=sys.stderr)
|
|
|
|
|
|
def ask_path(label: str, value: str | None = None, *, required: bool = True) -> Path | None:
|
|
if value is not None:
|
|
path = Path(value).expanduser().resolve()
|
|
if not path.is_file():
|
|
raise FileNotFoundError(f"Datei nicht gefunden: {path}")
|
|
return path
|
|
|
|
while True:
|
|
raw = input(f"{label}: ").strip()
|
|
if not raw and not required:
|
|
return None
|
|
if not raw:
|
|
print("Dieses Feld ist erforderlich.", file=sys.stderr)
|
|
continue
|
|
path = Path(raw).expanduser().resolve()
|
|
if path.is_file():
|
|
return path
|
|
print(f"Datei nicht gefunden: {path}", file=sys.stderr)
|
|
|
|
|
|
def confirm(question: str) -> bool:
|
|
return input(f"{question} [j/N]: ").strip().lower() in {"j", "ja", "y", "yes"}
|
|
|
|
|
|
def print_result(result: Any) -> None:
|
|
print(json.dumps(result, ensure_ascii=False, indent=2) if not isinstance(result, str) else result)
|
|
|
|
|
|
def _print_failure(error: BaseException) -> None:
|
|
print(f"Fehler: {type(error).__name__}: {error}", file=sys.stderr)
|
|
traceback.print_exception(type(error), error, error.__traceback__, file=sys.stderr)
|
|
|
|
|
|
def run(action: Callable[[], Any]) -> None:
|
|
"""Führt eine Werkzeugaktion aus und gibt Fehler samt vollständigem Stacktrace aus."""
|
|
try:
|
|
print_result(action())
|
|
except KeyboardInterrupt as error:
|
|
_print_failure(error)
|
|
raise SystemExit(130) from error
|
|
except Exception as error:
|
|
_print_failure(error)
|
|
raise SystemExit(1) from error
|