Orvanta Hub
Log in

Call an existing service in Python, normalise in TypeScript, and audit the result

Read from a service that is not going to be rewritten, reshape its response into the record your systems expect, and leave an audit trail — in one run, across two languages. ## Why this shape The blocker for most automation is rarely the new code. It is the service in a language nobody wants to touch, with a payload shape nobody wants to own. The usual answer is a migration project that never gets funded. This template is the other answer: call it as it is. The Python step talks to the legacy service because that is where the existing knowledge lives. The TypeScript step defines the record shape because that is where the rest of your estate lives. They are steps in the same flow, in the same run, sharing state. ## Steps 1. **fetch_legacy** (Python, stdlib only) — calls the service and returns the raw response. Non-JSON responses are passed through as text rather than crashing, because old services are not always well behaved. 2. **normalise** (TypeScript) — reshapes into a typed record and fails loudly on anything unexpected. **This is the step you are meant to edit.** 3. **record_audit** (TypeScript) — appends an audit entry over HTTP. ## What you need before running it - A reachable HTTP endpoint returning JSON. Any public test API works for a first run — the point is to see two runtimes hand state to each other. - Optionally an audit endpoint. Leave `audit_url` empty and the step reports that it did nothing, rather than pretending it succeeded. ## Adapting it - **Not HTTP?** Replace only `fetch_legacy`. SOAP, a database, a file drop and an SFTP poll all fit here; nothing downstream changes. - **Different field names:** `id_field`, `status_field` and `amount_field` are inputs, so the common case needs no code change at all. - **Audit to a table:** replace `record_audit`. It is the only step that knows where audit records go. - **Add a human gate:** see the triage template — the approval step drops in between `normalise` and any write.

json · 166 lines
1
{
2
  "summary": "Call an existing service in Python, normalise in TypeScript, and audit the result",
3
  "description": "Two runtimes in one run: Python reaches the legacy service, TypeScript owns the record shape.",
4
  "schema": {
5
    "$schema": "https://json-schema.org/draft/2020-12/schema",
6
    "type": "object",
7
    "order": [
8
      "base_url",
9
      "path",
10
      "api_key",
11
      "id_field",
12
      "status_field",
13
      "amount_field",
14
      "audit_url",
15
      "auth_header"
16
    ],
17
    "required": [
18
      "base_url",
19
      "path"
20
    ],
21
    "properties": {
22
      "base_url": {
23
        "type": "string",
24
        "description": "Base URL of the existing service.",
25
        "default": "https://jsonplaceholder.typicode.com"
26
      },
27
      "path": {
28
        "type": "string",
29
        "description": "Path to call on that service.",
30
        "default": "/todos/1"
31
      },
32
      "api_key": {
33
        "type": "string",
34
        "description": "Optional bearer token for the legacy service.",
35
        "default": ""
36
      },
37
      "id_field": {
38
        "type": "string",
39
        "description": "Field in the response holding the record id.",
40
        "default": "id"
41
      },
42
      "status_field": {
43
        "type": "string",
44
        "description": "Field holding the status.",
45
        "default": "title"
46
      },
47
      "amount_field": {
48
        "type": "string",
49
        "description": "Field holding a numeric amount, if there is one.",
50
        "default": "userId"
51
      },
52
      "audit_url": {
53
        "type": "string",
54
        "description": "Where audit entries are POSTed. Leave empty to skip auditing.",
55
        "default": ""
56
      },
57
      "auth_header": {
58
        "type": "string",
59
        "description": "Optional Authorization header for the audit endpoint.",
60
        "default": ""
61
      }
62
    }
63
  },
64
  "value": {
65
    "modules": [
66
      {
67
        "id": "fetch_legacy",
68
        "summary": "Call the existing service (Python)",
69
        "value": {
70
          "type": "rawscript",
71
          "tag": "",
72
          "language": "python3",
73
          "content": "import json\nimport urllib.request\nimport urllib.error\n\n\ndef main(\n    base_url: str,\n    path: str,\n    api_key: str = \"\",\n    timeout_seconds: int = 30,\n) -> dict:\n    \"\"\"Call an existing internal service and return its raw response.\n\n    Deliberately stdlib-only. This step stands in for the system nobody is\n    going to rewrite: it speaks whatever it speaks, and the flow adapts to it\n    rather than the other way round.\n\n    The raw payload is returned unmodified. Normalisation happens in the next\n    step, in a different language, which is the whole point of the template.\n    \"\"\"\n    url = base_url.rstrip(\"/\") + \"/\" + path.lstrip(\"/\")\n\n    request = urllib.request.Request(url, method=\"GET\")\n    request.add_header(\"Accept\", \"application/json\")\n    if api_key:\n        request.add_header(\"Authorization\", \"Bearer \" + api_key)\n\n    try:\n        with urllib.request.urlopen(request, timeout=timeout_seconds) as response:\n            body = response.read().decode(\"utf-8\")\n            status = response.status\n    except urllib.error.HTTPError as e:\n        # Surface the body: legacy services often explain themselves there and\n        # nowhere else.\n        raise RuntimeError(\n            \"Legacy service returned {}: {}\".format(e.code, e.read().decode(\"utf-8\", \"replace\")[:500])\n        )\n    except urllib.error.URLError as e:\n        raise RuntimeError(\"Could not reach legacy service at {}: {}\".format(url, e.reason))\n\n    try:\n        payload = json.loads(body)\n    except json.JSONDecodeError:\n        # Not everything old speaks JSON. Hand the text on and let the next\n        # step decide what to do with it.\n        payload = {\"_raw_text\": body}\n\n    return {\"status\": status, \"payload\": payload}\n",
74
          "input_transforms": {
75
            "base_url": {
76
              "type": "javascript",
77
              "expr": "flow_input.base_url"
78
            },
79
            "path": {
80
              "type": "javascript",
81
              "expr": "flow_input.path"
82
            },
83
            "api_key": {
84
              "type": "javascript",
85
              "expr": "flow_input.api_key"
86
            },
87
            "timeout_seconds": {
88
              "type": "static",
89
              "value": 30
90
            }
91
          }
92
        },
93
        "timeout": 60,
94
        "retry": {
95
          "exponential": {
96
            "attempts": 3,
97
            "multiplier": 2,
98
            "seconds": 2
99
          }
100
        }
101
      },
102
      {
103
        "id": "normalise",
104
        "summary": "Reshape into the record your systems expect (TypeScript)",
105
        "value": {
106
          "type": "rawscript",
107
          "tag": "",
108
          "language": "bun",
109
          "content": "type Normalised = {\n  id: string\n  status: string\n  amount: number | null\n  received_at: string\n  source: Record<string, unknown>\n}\n\n/**\n * Reshape whatever the legacy service returned into the record your systems\n * actually want.\n *\n * This is where the adapter logic lives, and it is the file you are expected to\n * edit. Everything else in this template is plumbing.\n */\nexport async function main(\n  response: { status: number; payload: Record<string, unknown> },\n  id_field = \"id\",\n  status_field = \"status\",\n  amount_field = \"amount\"\n): Promise<Normalised> {\n  const p = response.payload ?? {}\n\n  if (\"_raw_text\" in p) {\n    throw new Error(\n      \"Legacy service returned non-JSON. Parse it in this step before normalising: \" +\n        String(p._raw_text).slice(0, 200)\n    )\n  }\n\n  const id = p[id_field]\n  if (id === undefined || id === null || id === \"\") {\n    throw new Error(`Response has no '${id_field}' — check id_field against the real payload`)\n  }\n\n  const rawAmount = p[amount_field]\n  const amount =\n    rawAmount === undefined || rawAmount === null ? null : Number(rawAmount)\n  if (amount !== null && Number.isNaN(amount)) {\n    throw new Error(`'${amount_field}' is not numeric: ${JSON.stringify(rawAmount)}`)\n  }\n\n  return {\n    id: String(id),\n    status: String(p[status_field] ?? \"unknown\"),\n    amount,\n    received_at: new Date().toISOString(),\n    source: p\n  }\n}\n",
110
          "input_transforms": {
111
            "response": {
112
              "type": "javascript",
113
              "expr": "results.fetch_legacy"
114
            },
115
            "id_field": {
116
              "type": "javascript",
117
              "expr": "flow_input.id_field"
118
            },
119
            "status_field": {
120
              "type": "javascript",
121
              "expr": "flow_input.status_field"
122
            },
123
            "amount_field": {
124
              "type": "javascript",
125
              "expr": "flow_input.amount_field"
126
            }
127
          }
128
        }
129
      },
130
      {
131
        "id": "record_audit",
132
        "summary": "Append an audit entry",
133
        "value": {
134
          "type": "rawscript",
135
          "tag": "",
136
          "language": "bun",
137
          "content": "/**\n * Append an audit entry.\n *\n * Sent over HTTP so this template needs no database driver and no lockfile.\n * If your audit trail is a table rather than an endpoint, replace this step —\n * it is intentionally the only one that knows where audit records go.\n */\nexport async function main(\n  record: Record<string, unknown>,\n  audit_url: string,\n  flow_run_id: string,\n  auth_header = \"\"\n) {\n  if (!audit_url) {\n    // No audit endpoint configured: say so in the run output rather than\n    // silently succeeding. A silent no-op in an audit step is worse than a\n    // missing one, because it looks like coverage.\n    return { audited: false, reason: \"no audit_url configured\" }\n  }\n\n  const headers: Record<string, string> = { \"Content-Type\": \"application/json\" }\n  if (auth_header) headers[\"Authorization\"] = auth_header\n\n  const res = await fetch(audit_url, {\n    method: \"POST\",\n    headers,\n    body: JSON.stringify({\n      kind: \"legacy_sync\",\n      flow_run_id,\n      record_id: record.id,\n      status: record.status,\n      amount: record.amount,\n      observed_at: record.received_at\n    })\n  })\n\n  if (!res.ok) {\n    throw new Error(`Audit write failed: ${res.status} ${await res.text()}`)\n  }\n\n  return { audited: true, status: res.status }\n}\n",
138
          "input_transforms": {
139
            "record": {
140
              "type": "javascript",
141
              "expr": "results.normalise"
142
            },
143
            "audit_url": {
144
              "type": "javascript",
145
              "expr": "flow_input.audit_url"
146
            },
147
            "auth_header": {
148
              "type": "javascript",
149
              "expr": "flow_input.auth_header"
150
            },
151
            "flow_run_id": {
152
              "type": "javascript",
153
              "expr": "flow_input.flow_id ?? \"\""
154
            }
155
          }
156
        },
157
        "retry": {
158
          "constant": {
159
            "attempts": 2,
160
            "seconds": 3
161
          }
162
        }
163
      }
164
    ]
165
  }
166
}
Orvanta Hub — community scripts, flows, apps & resource types.