#!/usr/bin/env python3 """Apply padas-quickstart-core-api.json to Core (streams, connectors, task, ordered /start). Before running this script, start padas_tcp_quickstart_server.py and leave it running (see Getting Started → Quickstart: Core only). The sink connector expects something listening on the sink port when it starts. CLI arguments: --core-url Core API base URL (e.g. https://127.0.0.1:8999 or http://127.0.0.1:8999) --token Core API Bearer token (service account secret) Bundle file (first existing path wins, next to this script): padas-quickstart-core-api.json config/padas-quickstart-core-api.json ../config/padas-quickstart-core-api.json HTTPS: certificate verification is attempted first; if that fails, the request is retried once without verification (typical lab/self-signed Core). Use http:// if you prefer no TLS. """ from __future__ import annotations import argparse import http.client import json import ssl import sys import time import urllib.error import urllib.request from pathlib import Path from typing import Any def resolve_bundle_path(script_dir: Path) -> tuple[Path | None, list[Path]]: tried: list[Path] = [] for rel in ( Path("padas-quickstart-core-api.json"), Path("config") / "padas-quickstart-core-api.json", Path("..") / "config" / "padas-quickstart-core-api.json", ): p = (script_dir / rel).resolve() tried.append(p) if p.is_file(): return p, tried return None, tried def _fail_transport(url: str, exc: BaseException) -> None: print(f"Request failed: {url}\n {type(exc).__name__}: {exc}", file=sys.stderr) raise SystemExit(1) def _do_request(url: str, method: str, body: dict | None, token: str) -> tuple[int, str]: """HTTPS: verify certs first; on TLS failure retry once without verification.""" def build_req() -> urllib.request.Request: data = None if body is None else json.dumps(body).encode("utf-8") headers: dict[str, str] = {} if body is not None: headers["Content-Type"] = "application/json" headers["Authorization"] = f"Bearer {token}" return urllib.request.Request(url, data=data, headers=headers, method=method) def once(ctx: ssl.SSLContext | None) -> tuple[int, str]: req = build_req() try: with urllib.request.urlopen(req, context=ctx, timeout=120) as resp: raw = resp.read().decode("utf-8", errors="replace") return resp.status, raw except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", errors="replace") return e.code, raw if not url.startswith("https://"): try: return once(None) except http.client.BadStatusLine as e: _fail_transport(url, e) except urllib.error.URLError as e: _fail_transport(url, e) try: return once(None) except ssl.SSLError: print("warning: TLS verify failed; retrying without certificate verification", file=sys.stderr) insecure = ssl.create_default_context() insecure.check_hostname = False insecure.verify_mode = ssl.CERT_NONE return once(insecure) except urllib.error.URLError as e: if isinstance(e.reason, ssl.SSLError): print("warning: TLS verify failed; retrying without certificate verification", file=sys.stderr) insecure = ssl.create_default_context() insecure.check_hostname = False insecure.verify_mode = ssl.CERT_NONE return once(insecure) _fail_transport(url, e) except http.client.BadStatusLine as e: _fail_transport(url, e) def _entity_block(payload: dict[str, Any]) -> dict[str, Any] | None: for key in ("connector", "task", "stream"): block = payload.get(key) if isinstance(block, dict): return block return None def _print_status_row( kind: str, entity_id: str, http_code: int, payload: dict[str, Any] | None, raw: str, ) -> None: if payload is None: err = raw[:200].replace("\n", " ") if raw else "" print(f"{kind:10} {entity_id:40} HTTP {http_code:3} (non-JSON or empty) {err}") return inner = _entity_block(payload) or payload status = inner.get("status", "?") last_err = inner.get("last_error") uptime = inner.get("uptime", "") en = inner.get("enabled") extra = [] if en is not None: extra.append(f"enabled={en}") if uptime: extra.append(f"uptime={uptime}") if last_err: extra.append(f"last_error={last_err!r}") tail = " ".join(extra) if extra else "" st = str(status).lower() healthy = st == "running" or (kind == "stream" and st == "active") flag = "" if healthy else " <--" print(f"{kind:10} {entity_id:40} HTTP {http_code:3} status={status!s:12} {tail}{flag}") def print_status_summary( base: str, token: str, streams: list[Any], connectors: list[Any], task: dict[str, Any] | None, ) -> None: time.sleep(0.75) print() print("--- status ---") print() def get(path: str) -> tuple[int, dict[str, Any] | None, str]: url = f"{base}{path}" code, text = _do_request(url, "GET", None, token) if code != 200: return code, None, text try: return code, json.loads(text), text except json.JSONDecodeError: return code, None, text for body in streams: if not isinstance(body, dict): continue sid = (body.get("stream") or {}).get("id") if not sid: continue code, data, raw = get(f"/api/v1/streams/{sid}/status") _print_status_row("stream", sid, code, data, raw) for body in connectors: if not isinstance(body, dict): continue cid = (body.get("connector") or {}).get("id") if not cid: continue code, data, raw = get(f"/api/v1/connectors/{cid}/status") _print_status_row("connector", cid, code, data, raw) if isinstance(task, dict): tid = (task.get("task") or {}).get("id") if tid: code, data, raw = get(f"/api/v1/tasks/{tid}/status") _print_status_row("task", tid, code, data, raw) print() def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--core-url", required=True, help="Core API base URL (e.g. https://127.0.0.1:8999)", ) parser.add_argument( "--token", required=True, help="Core API Bearer token (service account)", ) args = parser.parse_args() token = args.token.strip() if not token: print("--token must be non-empty", file=sys.stderr) raise SystemExit(1) script_dir = Path(__file__).resolve().parent bundle_path, tried = resolve_bundle_path(script_dir) if bundle_path is None: print("Could not find padas-quickstart-core-api.json. Tried:\n " + "\n ".join(str(p) for p in tried), file=sys.stderr) raise SystemExit(1) bundle = json.loads(bundle_path.read_text(encoding="utf-8")) streams = bundle.get("streams") or [] connectors = bundle.get("connectors") or [] task = bundle.get("task") starts = bundle.get("starts_in_order") or [] base = args.core_url.rstrip("/") def post(path: str, body: dict) -> None: url = f"{base}{path}" code, text = _do_request(url, "POST", body, token) if code not in (200, 201): print(f"HTTP {code} {url}\n{text[:4000]}", file=sys.stderr) if code == 409: print("Hint: object id may already exist; delete it or use a clean config_dir.", file=sys.stderr) raise SystemExit(1) def post_empty(path: str) -> None: url = f"{base}{path}" code, text = _do_request(url, "POST", None, token) if code not in (200, 201): print(f"HTTP {code} {url}\n{text[:4000]}", file=sys.stderr) raise SystemExit(1) for body in streams: post("/api/v1/streams", body) for body in connectors: post("/api/v1/connectors", body) if task: post("/api/v1/tasks", task) for item in starts: kind = item.get("kind") cid = item.get("id") if kind == "connector": post_empty(f"/api/v1/connectors/{cid}/start") elif kind == "task": post_empty(f"/api/v1/tasks/{cid}/start") else: print(f"Unknown start kind: {item!r}", file=sys.stderr) return 1 print_status_summary(base, token, streams, connectors, task) print("Done. Run padas_tcp_quickstart_server.py and watch for [SINK] lines (Quickstart: Core only).") return 0 if __name__ == "__main__": raise SystemExit(main())