#!/usr/bin/env python3 """ etlworks_api.py — Python starter for the Etlworks REST API. A small client with no dependencies beyond the Python 3.8+ standard library. Use it as a library or from the command line, and edit it freely: it is a starting point, not a package. Authentication, in order of precedence: ETLWORKS_API_KEY an API key (Settings > Users > edit user > API key) ETLWORKS_TOKEN a JWT you already have ETLWORKS_LOGIN and ETLWORKS_PASSWORD exchanged for a JWT via POST /v1/token-auth/issue (accounts with two-factor authentication need an API key instead) ETLWORKS_URL is the instance, for example https://app.etlworks.com (the client adds /rest). Usage as a library: from etlworks_api import EtlworksApi api = EtlworksApi("https://app.etlworks.com", api_key="your-key") for flow in api.list_flows(): print(flow["id"], flow["name"]) run = api.run_flow(123, params={"date": "2026-10-01"}) final = api.wait(123, run["auditId"]) # polls until it finishes print(final["status"]) # success, warning, error, canceled print(api.execution_log(123, run["auditId"])) for result in api.cli("list-flows select id, name limit 5"): print(result["stdout"]) api.request("GET", "/v1/schedules") # anything else in the API Usage from the command line: export ETLWORKS_URL="https://app.etlworks.com" export ETLWORKS_API_KEY="your-key" python etlworks_api.py flows python etlworks_api.py run 123 '{"date": "2026-10-01"}' --wait python etlworks_api.py run "Load orders" --wait python etlworks_api.py status 123 4567 python etlworks_api.py log 123 4567 python etlworks_api.py stop 123 4567 python etlworks_api.py history 123 python etlworks_api.py cli "list-flows select id, name limit 5" python etlworks_api.py request GET /v1/schedules python etlworks_api.py login # print a JWT for the login/password in the env `run --wait` exits 0 for success or warning and 1 for error or canceled, so it can gate a CI step. Reference: https://etlworks.com/dev/api/ """ import json import os import sys import time import urllib.error import urllib.parse import urllib.request TERMINAL = {"success", "warning", "error", "canceled"} STARTED = {0, 10} # 10: handed to an Integration Agent class EtlworksApiError(Exception): def __init__(self, status, body, method, path): self.status, self.body = status, body super().__init__(f"{method} {path} -> HTTP {status}: {body[:500] if body else '(empty body)'}") class EtlworksApi: """Client for the Etlworks REST API.""" def __init__(self, base_url=None, api_key=None, token=None, login=None, password=None, timeout=120): self.base_url = (base_url or os.environ.get("ETLWORKS_URL", "http://localhost:8080")).rstrip("/") if self.base_url.endswith("/rest"): self.base_url = self.base_url[:-5] self.timeout = timeout self.token = (api_key or os.environ.get("ETLWORKS_API_KEY") or token or os.environ.get("ETLWORKS_TOKEN")) if not self.token: login = login or os.environ.get("ETLWORKS_LOGIN") password = password or os.environ.get("ETLWORKS_PASSWORD") if not (login and password): raise ValueError("Set ETLWORKS_API_KEY, ETLWORKS_TOKEN, or ETLWORKS_LOGIN and ETLWORKS_PASSWORD.") self.token = self.issue_token(login, password) # -- transport ------------------------------------------------------- def request(self, method, path, body=None, params=None, auth=True, raw=False): """Call any endpoint. `path` starts with /v1/. Returns parsed JSON, text, or None.""" url = self.base_url + "/rest" + path if params: query = {k: str(v).lower() if isinstance(v, bool) else v for k, v in params.items() if v is not None} url += ("&" if "?" in url else "?") + urllib.parse.urlencode(query) data = None headers = {"Accept": "application/json, text/plain, */*"} if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" if auth: headers["Authorization"] = "Bearer " + self.token req = urllib.request.Request(url, data=data, method=method.upper(), headers=headers) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: payload = resp.read().decode("utf-8", errors="replace") if raw or not payload: return payload or None if "json" in (resp.headers.get("Content-Type") or ""): return json.loads(payload) return payload except urllib.error.HTTPError as e: raise EtlworksApiError(e.code, e.read().decode("utf-8", errors="replace"), method, path) from None def issue_token(self, login, password): """Exchange a login and password for a JWT (valid 8 hours by default).""" resp = self.request("POST", "/v1/token-auth/issue", {"login": login, "password": password}, auth=False) if not resp or not resp.get("token"): raise ValueError("No token returned. Accounts with two-factor authentication need an API key.") return resp["token"] # -- flows and runs -------------------------------------------------- def list_flows(self): return self.request("GET", "/v1/flows") def run_flow(self, flow, params=None, **options): """Start a flow by id (int) or name (str). `options` become query parameters, for example sync=True, allowParallel=True, emails="ops@example.com".""" if isinstance(flow, int) or str(flow).isdigit(): path = f"/v1/flows/{int(flow)}/run" else: path = "/v1/flows/flow-name/" + urllib.parse.quote(str(flow), safe="") + "/run" return self.request("POST", path, params or {}, params=options) def execution(self, flow_id, audit_id): return self.request("GET", f"/v1/executions/{int(flow_id)}", params={"auditId": int(audit_id)}) def wait(self, flow_id, audit_id, interval=5, timeout=None): """Poll until the execution reaches success, warning, error or canceled.""" start = time.time() while True: rec = self.execution(flow_id, audit_id) if rec and rec.get("status") in TERMINAL: return rec if timeout and time.time() - start > timeout: raise TimeoutError(f"execution {audit_id} still {rec.get('status') if rec else 'unknown'}") time.sleep(interval) def execution_log(self, flow_id, audit_id): return self.request("GET", f"/v1/executions/{int(flow_id)}/audit/{int(audit_id)}/console", raw=True) def stop(self, flow_id, audit_id=None): return self.request("POST", f"/v1/flows/{int(flow_id)}/stop", params={"auditId": audit_id}) def history(self, flow_id): return self.request("GET", f"/v1/executions/{int(flow_id)}/history") # -- CLI ------------------------------------------------------------- def cli(self, command, variables=None, tenant=None): """Run one or more built-in CLI commands. Returns one result per command.""" body = {"command": command} if variables: body["vars"] = variables if tenant: body["tenant"] = tenant return self.request("POST", "/v1/cli", body) # ---------------------------------------------------------------------- # Command line # ---------------------------------------------------------------------- def _print(value): if isinstance(value, (dict, list)): print(json.dumps(value, indent=2)) elif value is not None: print(value) def main(argv): if not argv or argv[0] in ("-h", "--help", "help"): print(__doc__) return 0 cmd, args = argv[0], argv[1:] wait = "--wait" in args args = [a for a in args if a != "--wait"] if cmd == "login": api = EtlworksApi.__new__(EtlworksApi) api.base_url = os.environ.get("ETLWORKS_URL", "http://localhost:8080").rstrip("/") api.timeout, api.token = 120, None print(api.issue_token(os.environ["ETLWORKS_LOGIN"], os.environ["ETLWORKS_PASSWORD"])) return 0 api = EtlworksApi() if cmd == "flows": for f in api.list_flows() or []: print(f"{f.get('id')}\t{f.get('name')}") elif cmd == "run": params = json.loads(args[1]) if len(args) > 1 else {} run = api.run_flow(args[0], params) if run.get("status") not in STARTED: _print(run) return 1 print(f"started: flowId={run['flowId']} auditId={run['auditId']}", file=sys.stderr) if not wait: _print(run) return 0 final = api.wait(run["flowId"], run["auditId"]) print(f"finished: {final['status']}", file=sys.stderr) if final["status"] in ("error", "canceled"): print(api.execution_log(run["flowId"], run["auditId"]) or "", file=sys.stderr) return 1 elif cmd == "status": _print(api.execution(args[0], args[1])) elif cmd == "log": _print(api.execution_log(args[0], args[1])) elif cmd == "stop": _print(api.stop(args[0], args[1] if len(args) > 1 else None)) elif cmd == "history": _print(api.history(args[0])) elif cmd == "cli": results = api.cli(args[0]) worst = 0 for r in results or []: if r.get("stdout"): print(r["stdout"]) if r.get("stderr"): print(r["stderr"], file=sys.stderr) worst = max(worst, r.get("exitCode") or 0) return worst elif cmd == "request": body = json.loads(args[2]) if len(args) > 2 else None _print(api.request(args[0], args[1], body)) else: print(f"Unknown command: {cmd}. Run with --help.", file=sys.stderr) return 2 return 0 if __name__ == "__main__": try: sys.exit(main(sys.argv[1:])) except (EtlworksApiError, ValueError, TimeoutError) as e: print(f"error: {e}", file=sys.stderr) sys.exit(1)