translating-workflows

name	translating-workflows
description	Converts analyzed n8n workflow subgraphs into pure Python/FastAPI code. Handles large workflows via sequential subgraph processing. Generates routers, services, models, and mocks. Activate when credentials are collected and the user is ready to translate, convert, or generate code from their n8n workflows.

Translating Workflows

You are converting n8n workflows into production-grade Python/FastAPI code. You process one subgraph at a time to stay within context limits. You present a plan for each file before writing it.

---

0. Critical Rules — Read First

These are the failure modes that have cost the most time on real migrations. Internalize them BEFORE writing any code.

0.1 — asyncpg binds parameters BEFORE the SQL parses

This is the single most expensive bug class in n8n→Python migration.

asyncpg does NOT honor SQL casts like $1::time, $2::date, or $3::double precision when you pass strings. The driver binds the Python value to the wire protocol before PostgreSQL ever sees the SQL, so the cast cannot rescue you.

Wrong — fails with 'str' object has no attribute 'hour':

await conn.execute(
    "INSERT INTO prayer_cache (user_id, fajr) VALUES ($1, $2::time)",
    user_id, "05:30"
)

Right — convert in Python first:

from datetime import time
hh, mm = "05:30".split(":")
await conn.execute(
    "INSERT INTO prayer_cache (user_id, fajr) VALUES ($1, $2)",
    user_id, time(int(hh), int(mm))
)

The same applies to:

• $X::date → pass datetime.date(y, m, d), not "2026-04-09"
• $X::double precision → pass float(value), not "30.0444"
• $X::int → pass int(value), not "123"
• $X::jsonb → pass json.dumps(obj) (this one DOES work — jsonb cast accepts text)

Scan for this when reviewing translation output. Every ::time, ::date, ::float, ::int, ::numeric cast in SQL is a smoke signal. Replace with explicit Python conversion.

0.2 — NEVER paraphrase Code-node content

When porting an n8n Code node (especially long ones with multilingual strings), extract the parameters.jsCode verbatim and port it byte-for-byte. Do NOT:

• Summarize message pools or "shorten for clarity"
• Translate Arabic/CJK strings to English
• Skip scenario branches because "the gist is the same"
• Reduce an 800-line node to a 35-line port

The user is migrating because they want identical behavior. Loss of fidelity in any string-producing node will be visible to end users immediately. For Code nodes larger than ~50 lines, plan to manually re-port byte-for-byte instead of trusting the first-pass agent output.

0.3 — Inter-workflow "Call X" nodes are FUNCTION CALLS, not HTTP POSTs

When workflow A's "Execute Workflow" node points at workflow B (or workflow A HTTP-POSTs to a webhook owned by workflow B in the same n8n instance), and BOTH workflows are being translated together, replace the inter-workflow call with a direct in-process Python function call. Do NOT:

• Generate httpx.post("https://n8n.example.com/webhook/settings-menu", ...)
• Hardcode the production webhook URL anywhere

The inter-workflow HTTP call would silently send dev traffic to the production n8n instance, returning production data and corrupting tests.

0.4 — Webhook & WebApp URLs come from settings.PUBLIC_BASE_URL

Every URL that appears in outgoing messages, inline-keyboard web_app buttons, callback URLs, etc., MUST be built from settings.PUBLIC_BASE_URL. Hardcoded production domains make dev/staging/prod indistinguishable and cause cross- environment data leaks — dev users have ended up viewing production dashboards and dev callbacks have ended up writing to production rows because of this.

0.5 — Disabled n8n nodes default to skipped

Before translating any node, check node.disabled === true in the workflow JSON. If a node (or an entire chain entered via a disabled trigger) is disabled, skip it by default and log it in a "deliberate divergences" report:

Skipped (disabled in n8n): "Daily 4am Check" → entire inactive cleanup chain
  Re-enable in dev? (yes/no, default: no)

The user must explicitly opt back in. Translating a disabled node enables behavior the user has deliberately turned off in production.

0.6 — Strip None from outgoing JSON to third-party APIs

Many APIs (Telegram, OpenRouter, Aladhan, Slack, ...) reject null for optional fields with cryptic 400 errors (e.g., Bad Request: unsupported parse_mode). Every translated HTTP client wrapper should drop None keys before sending:

async def api_call(method: str, **kwargs) -> dict:
    payload = {k: v for k, v in kwargs.items() if v is not None}
    response = await client.post(url, json=payload)
    if response.status_code >= 400:
        logger.error("API error %s: %s", response.status_code, response.text)
    response.raise_for_status()
    return response.json()

0.7 — LangChain Postgres chat memory has a JSONB schema, not (role, content)

The @n8n/n8n-nodes-langchain.memoryPostgresChat node uses LangChain's table shape, NOT a naive (role, content) schema:

CREATE TABLE n8n_chat_histories (
    id           SERIAL PRIMARY KEY,
    session_id   TEXT NOT NULL,
    message      JSONB NOT NULL
);

message is {"type": "human" | "ai", "data": {"content": "..."}}. Inserts must match this exact shape:

await conn.execute(
    "INSERT INTO n8n_chat_histories (session_id, message) VALUES ($1, $2::jsonb)",
    str(telegram_id),
    json.dumps({"type": "human", "data": {"content": user_text}})
)

Reads must unwrap message["data"]["content"]. Every translated chat workflow must honor this exact schema or messages will be invisible to LangChain readers.

0.8 — Validate surrogate pairs when fixing emoji escapes

When converting \uXXXX\uXXXX JS-style escapes to Python literals, the resulting codepoint MUST match the intended emoji visually. We have already shipped \ud83e\uddf2 (folded hands U+1F932) corrupted to U+1F9F2 (magnet — looks similar in small fonts but completely changes meaning). After conversion, eyeball every emoji at normal size, not just in the editor.

0.9 — JSON_AGG / JSONB columns return strings from asyncpg, not parsed dicts

asyncpg returns jsonb columns as Python str, not dict/list. Always json.loads() them before returning to clients or iterating. Symptom: a JS client doing [...calendar_data].sort(...) spreads the string character-by- character, producing NaN or garbled output.

row = await conn.fetchrow("SELECT JSON_AGG(...) AS data FROM ...")
data = json.loads(row["data"]) if row["data"] else []

---

1. Pre-Translation Setup

1a. Initialize the Application Skeleton

STOP and present this plan to the user before creating anything:

I will create the following base files:
  workspace/src/main.py          — FastAPI app entry point
  workspace/src/config.py        — pydantic-settings configuration
  workspace/src/database.py      — asyncpg connection pool
  workspace/requirements.txt     — Python dependencies
  workspace/Dockerfile           — Container definition
  workspace/docker-compose.yml   — Service orchestration
  workspace/.env.example         — Copy of config/.env.example

Approve? (yes/no)

1b. Base Dependencies

fastapi>=0.115.0
uvicorn[standard]>=0.30.0
pydantic>=2.0
pydantic-settings>=2.0
httpx>=0.27.0
asyncpg>=0.30.0
apscheduler>=3.10.0
backoff>=2.2.0
pytest>=8.0
pytest-asyncio>=0.24.0

Add ONLY what is needed. If a workflow does not use scheduling, do NOT include apscheduler. If it does not use Postgres, do NOT include asyncpg.

---

2. Translation Procedure

Processing Order

1. Base application skeleton (main.py, config.py, database.py)
2. Workflows: simplest first, most complex last
3. Within each workflow: subgraphs in topological order

For Each Subgraph

Step 1 — Load only this subgraph's nodes:

# Extract specific nodes belonging to this subgraph
cat migration-state/discovery/workflows/<wf_id>.json | jq '
  .nodes[] | select(.name == "Node A" or .name == "Node B" or ...)'

# Extract connections between these nodes only
cat migration-state/discovery/workflows/<wf_id>.json | jq '
  .connections | to_entries[] |
  select(.key == "Node A" or .key == "Node B") |
  {(.key): .value}'

Do NOT load the entire workflow JSON. Only load what you need for this subgraph.

Step 2 — Check for unknown patterns:

Before writing code, verify every node in this subgraph has a known translation. Reference migration-state/learned-nodes.json. If any node lacks a mapping, STOP and resolve it using the analyzing-workflows skill before continuing.

Step 3 — Present the file plan:

Subgraph: [name] ([n] nodes)
Workflow: [workflow_name]

Files I will create/modify:
  CREATE  workspace/src/routers/[workflow]_router.py     — Route handlers
  CREATE  workspace/src/services/[workflow]_[subgraph].py — Business logic
  CREATE  workspace/src/models/[workflow]_models.py       — Pydantic models
  MODIFY  workspace/src/main.py                           — Register router

Functions to generate:
  handle_[trigger_name]()     → route handler (from: Webhook node)
  process_[node_name]()       → service function (from: IF node)
  fetch_[node_name]()         → HTTP call (from: HTTP Request node)
  query_[node_name]()         → DB query (from: Postgres node)

Approve? (yes/no)

Step 4 — Translate node by node:

Walk the subgraph in topological order. For each node, apply the translation patterns from .claude/skills/translating-workflows/references/translation-patterns.md.

Key rules:

• One function per node. Function name derived from sanitized node name.
• Every function signature: async def <name>(items: list[dict]) -> list[dict]
• Inter-function data validated by Pydantic models at boundaries.
• All HTTP calls wrapped with @backoff.on_exception(backoff.expo, httpx.HTTPError, max_tries=3)
• All DB queries use parameterized statements. NEVER string-format user data into SQL.
• All credentials from config.settings. NEVER hardcoded.

Step 5 — Handle side effects in dev mode:

For any function that has side effects (sends messages, writes external data):

async def send_telegram_message(items: list[dict]) -> list[dict]:
    if settings.DRY_RUN:
        logger.info(f"[DRY RUN] Would send to Telegram: {items}")
        return items

    # actual implementation
    ...

Add DRY_RUN: bool = True to the settings model. Default to True. The user must explicitly set DRY_RUN=false to enable real side effects.

Step 6 — Handle mocked integrations:

For integrations with strategy "mock" in the registry, generate mock modules:

# workspace/src/mocks/stripe_mock.py

class StripeMock:
    async def create_charge(self, amount: int, currency: str) -> dict:
        return {
            "id": "ch_mock_123",
            "amount": amount,
            "currency": currency,
            "status": "succeeded"
        }

Wire mocks via dependency injection in FastAPI so they can be swapped for real clients without code changes.

Step 7 — Update the manifest:

After each subgraph is translated, update migration-state/manifest.json:

{
  "subgraphs": [
    {"id": "sg_a", "status": "translated", "files": ["router.py", "service.py"]},
    {"id": "sg_b", "status": "pending"}
  ]
}

---

3. Expression Translation

n8n uses ={{ expression }} syntax. Translate carefully — these are the #1 source of silent bugs.

n8n Expression	Python Equivalent
{{ $json.fieldName }}	item["fieldName"]
{{ $json["field"]["nested"] }}	item["field"]["nested"]
{{ $node["Node Name"].json.field }}	Return value from that node's function
{{ $env.VAR_NAME }}	settings.VAR_NAME
{{ $now }}	datetime.now(timezone.utc)
{{ $runIndex }}	Loop index variable
{{ $input.item.json.field }}	item["field"]
{{ $json.field.toString() }}	str(item["field"])
{{ $json.items.length }}	len(item["items"])

ALWAYS flag translated expressions with an inline comment:

# TRANSLATED FROM: {{ $json.prayer_count + 1 }}
prayer_count = item["prayer_count"] + 1

This makes review easy and bugs traceable.

---

4. Mid-Chain Webhook Responses

In n8n, "Respond to Webhook" can appear anywhere in the flow. In FastAPI, the response MUST be returned from the route handler.

If the workflow responds mid-chain:

@router.post("/webhook/prayer")
async def handle_prayer_webhook(request: Request, background_tasks: BackgroundTasks):
    payload = await request.json()

    # Immediate validation and response (was: Respond to Webhook node)
    if not validate_payload(payload):
        return JSONResponse(status_code=400, content={"error": "Invalid"})

    # Queue remaining processing as background task
    background_tasks.add_task(process_prayer_chain, payload)

    # This is what the user sees immediately
    return JSONResponse(status_code=200, content={"status": "accepted"})


async def process_prayer_chain(payload: dict):
    # All nodes AFTER the Respond to Webhook node run here
    ...

---

5. Completion Criteria

A workflow is fully translated when:

• All subgraphs have status: translated in the manifest
• All files follow the project structure conventions
• All credentials come from environment variables
• All side-effect functions have DRY_RUN guards
• All mocked integrations use dependency injection
• No file exceeds 300 lines

Present to user: "Workflow [name] is fully translated. Ready to generate tests (Phase 5)?"

---

6. Known Edge Cases

Patterns that have already burned at least one migration. Check for them explicitly during translation review.

1. Friday Jumuah for Dhuhr. Some prayer/scheduling logic has a special branch for prayer == 'Dhuhr' && isFriday that uses a separate JUMUAH message pool AND skips other branches (e.g. continue in the JS loop). Easy to miss when porting nested loops — re-read the JS top-to-bottom to confirm every continue and break is honored.

2. Period mode disables prayer actions for female users. Logic like period_mode_active && gender === 'female' disables notifications, callbacks, and azkar atomically. Both the business logic AND any LLM system prompt have to honor this — easy to miss the prompt side.

3. First-day timezone-safe SQL check. Patterns like users.created_at::date = (NOW() AT TIME ZONE tz)::date belong in SQL, not in Python — moving the comparison to Python's date.today() introduces a timezone bug that only fails for users in non-UTC timezones around midnight.

4. EXTRACT(DAY FROM interval) returns a Postgres numeric that asyncpg surfaces as Decimal/str. Comparing int > 2 against it raises TypeError. Coerce explicitly: int(user.get("days_inactive") or 0).

5. Telegram editMessageText 400 "unsupported parse_mode". Caused by sending "parse_mode": null in JSON. See §0.6 — strip None from outgoing kwargs.

6. Webapp inline keyboard web_app.url MUST be HTTPS and absolute. Relative paths fail silently. Build from settings.PUBLIC_BASE_URL, never hardcode.

7. asyncpg time/date/float codec bugs in 5+ places. See §0.1.

8. is_paused AND is_onboarded are TWO different filters. Many heartbeat queries need WHERE is_onboarded AND NOT is_paused. Forgetting one ships messages to paused users or skips active ones.

9. Telegram callback data length is capped at 64 bytes. Composite payloads like action|prayer|yes|2026-04-09|7946980274|123456789 can sit right at the limit; longer prayer names or extra fields will silently break the keyboard. Validate length at construction time.

10. Container TZ vs APScheduler TZ vs Postgres TZ — three layers. Setting -e TZ=Africa/Cairo on the container handles container time. APScheduler CronTriggers additionally need timezone="Africa/Cairo" (or whatever n8n ran in) — APScheduler defaults to UTC even inside a Cairo container. Postgres NOW() AT TIME ZONE 'Africa/Cairo' is independent of both. All three must agree or scheduled jobs and "today" comparisons will drift.