n8n-code-python

Python 代码节点（Beta）

在 n8n Code 节点中编写 Python 代码的专业指南。

---

⚠️ 重要提示：优先使用 JavaScript

建议：95% 的使用场景都应使用 JavaScript。仅在以下情况使用 Python：

• 需要特定的 Python 标准库函数
• 你对 Python 语法明显更熟悉
• 你正在进行更适合使用 Python 的数据转换

优先使用 JavaScript 的原因：

• 提供完整的 n8n 辅助函数（$helpers.httpRequest 等）
• Luxon DateTime 库支持高级日期/时间操作
• 不受外部库限制
• n8n 文档和社区支持更完善

---

快速开始

# Basic template for Python Code nodes
items = _input.all()

# Process data
processed = []
for item in items:
    processed.append({
        "json": {
            **item["json"],
            "processed": True,
            "timestamp": datetime.now().isoformat()
        }
    })

return processed

基本规则

1. 先考虑 JavaScript——仅在必要时使用 Python
2. 访问数据：_input.all()、_input.first() 或 _input.item
3. 重要：必须返回 [{"json": {...}}] 格式
4. 重要：Webhook 数据位于 _json["body"] 中（而不是直接位于 _json 中）
5. 重要限制：不能使用外部库（不支持 requests、pandas、numpy）
6. 仅限标准库：json、datetime、re、base64、hashlib、urllib.parse、math、random、statistics

---

模式选择指南

与 JavaScript 相同——根据使用场景选择：

对所有条目运行一次（推荐 - 默认）

适用场景： 95% 的使用场景

• 工作方式：无论输入数量多少，代码都只执行一次
• 数据访问：_input.all() 或 _items 数组（Native 模式）
• 最适合：聚合、筛选、批处理、转换
• 性能：处理多个条目时更快（只执行一次）

# Example: Calculate total from all items
all_items = _input.all()
total = sum(item["json"].get("amount", 0) for item in all_items)

return [{
    "json": {
        "total": total,
        "count": len(all_items),
        "average": total / len(all_items) if all_items else 0
    }
}]

对每个条目运行一次

适用场景： 仅适用于特殊情况

• 工作方式：针对每个输入条目分别执行代码
• 数据访问：_input.item 或 _item（Native 模式）
• 最适合：针对单个条目的逻辑、独立操作、逐条验证
• 性能：处理大型数据集时更慢（需要多次执行）

# Example: Add processing timestamp to each item
item = _input.item

return [{
    "json": {
        **item["json"],
        "processed": True,
        "processed_at": datetime.now().isoformat()
    }
}]

---

Python 模式：Beta 与 Native

n8n 提供两种 Python 执行模式：

Python（Beta）- 推荐

• 使用：_input、_json、_node 辅助语法
• 最适合：大多数 Python 使用场景
• 可用辅助函数：_now、_today、_jmespath()
• 导入：from datetime import datetime

# Python (Beta) example
items = _input.all()
now = _now  # Built-in datetime object

return [{
    "json": {
        "count": len(items),
        "timestamp": now.isoformat()
    }
}]

Python（Native）（Beta）

• 使用：仅使用 _items、_item 变量
• 无辅助函数：不支持 _input、_now 等
• 功能更有限：仅限标准 Python
• 适用场景：需要不使用 n8n 辅助函数的纯 Python

# Python (Native) example
processed = []

for item in _items:
    processed.append({
        "json": {
            "id": item["json"].get("id"),
            "processed": True
        }
    })

return processed

建议：使用 Python（Beta），以获得更好的 n8n 集成体验。

---

数据访问模式

模式 1：_input.all() - 最常见

适用场景：处理数组、批量操作、聚合

# Get all items from previous node
all_items = _input.all()

# Filter, transform as needed
valid = [item for item in all_items if item["json"].get("status") == "active"]

processed = []
for item in valid:
    processed.append({
        "json": {
            "id": item["json"]["id"],
            "name": item["json"]["name"]
        }
    })

return processed

模式 2：_input.first() - 很常见

适用场景：处理单个对象、API 响应

# Get first item only
first_item = _input.first()
data = first_item["json"]

return [{
    "json": {
        "result": process_data(data),
        "processed_at": datetime.now().isoformat()
    }
}]

模式 3：_input.item - 仅适用于逐条运行模式

适用场景：处于“对每个条目运行一次”模式时

# Current item in loop (Each Item mode only)
current_item = _input.item

return [{
    "json": {
        **current_item["json"],
        "item_processed": True
    }
}]

模式 4：_node - 引用其他节点

适用场景：需要工作流中特定节点的数据

# Get output from specific node
webhook_data = _node["Webhook"]["json"]
http_data = _node["HTTP Request"]["json"]

return [{
    "json": {
        "combined": {
            "webhook": webhook_data,
            "api": http_data
        }
    }
}]

参见：DATA_ACCESS.md，了解全面的指南

---

重要：Webhook 数据结构

最常见的错误：Webhook 数据嵌套在 ["body"] 下

# ❌ WRONG - Will raise KeyError
name = _json["name"]
email = _json["email"]

# ✅ CORRECT - Webhook data is under ["body"]
name = _json["body"]["name"]
email = _json["body"]["email"]

# ✅ SAFER - Use .get() for safe access
webhook_data = _json.get("body", {})
name = webhook_data.get("name")

原因：Webhook 节点会将所有请求数据封装在 body 属性下，包括 POST 数据、查询参数和 JSON 负载。

参见：DATA_ACCESS.md，了解完整的 Webhook 结构详情

---

返回格式要求

重要规则：始终返回包含 "json" 键的字典列表

正确的返回格式

# ✅ Single result
return [{
    "json": {
        "field1": value1,
        "field2": value2
    }
}]

# ✅ Multiple results
return [
    {"json": {"id": 1, "data": "first"}},
    {"json": {"id": 2, "data": "second"}}
]

# ✅ List comprehension
transformed = [
    {"json": {"id": item["json"]["id"], "processed": True}}
    for item in _input.all()
    if item["json"].get("valid")
]
return transformed

# ✅ Empty result (when no data to return)
return []

# ✅ Conditional return
if should_process:
    return [{"json": processed_data}]
else:
    return []

错误的返回格式

# ❌ WRONG: Dictionary without list wrapper
return {
    "json": {"field": value}
}

# ❌ WRONG: List without json wrapper
return [{"field": value}]

# ❌ WRONG: Plain string
return "processed"

# ❌ WRONG: Incomplete structure
return [{"data": value}]  # Should be {"json": value}

原因：下游节点需要列表格式。格式不正确会导致工作流执行失败。

参见：ERROR_PATTERNS.md #2，了解详细的错误解决方案

---

重要限制：不能使用外部库

Python 最重要的限制：不能导入外部包

不可用的内容

# ❌ NOT AVAILABLE - Will raise ModuleNotFoundError
import requests  # ❌ No
import pandas  # ❌ No
import numpy  # ❌ No
import scipy  # ❌ No
from bs4 import BeautifulSoup  # ❌ No
import lxml  # ❌ No

可用内容（标准库）

# ✅ AVAILABLE - Standard library only
import json  # ✅ JSON parsing
import datetime  # ✅ Date/time operations
import re  # ✅ Regular expressions
import base64  # ✅ Base64 encoding/decoding
import hashlib  # ✅ Hashing functions
import urllib.parse  # ✅ URL parsing
import math  # ✅ Math functions
import random  # ✅ Random numbers
import statistics  # ✅ Statistical functions

替代方案

需要发送 HTTP 请求？

• ✅ 在 Code 节点之前使用 HTTP Request 节点
• ✅ 或切换到 JavaScript 并使用 $helpers.httpRequest()

需要进行数据分析（pandas/numpy）？

• ✅ 使用 Python statistics 模块进行基本统计
• ✅ 或切换到 JavaScript 执行大多数操作
• ✅ 使用列表和字典进行手动计算

需要网页抓取（BeautifulSoup）？

• ✅ 使用 HTTP Request 节点 + HTML Extract 节点
• ✅ 或切换到 JavaScript，使用正则表达式/字符串方法

参见：STANDARD_LIBRARY.md，了解完整参考

---

常见模式概览

根据生产工作流，以下是最实用的 Python 模式：

1. 数据转换

使用列表推导式转换所有项目

items = _input.all()

return [
    {
        "json": {
            "id": item["json"].get("id"),
            "name": item["json"].get("name", "Unknown").upper(),
            "processed": True
        }
    }
    for item in items
]

2. 筛选与聚合

使用内置函数求和、筛选和计数

items = _input.all()
total = sum(item["json"].get("amount", 0) for item in items)
valid_items = [item for item in items if item["json"].get("amount", 0) > 0]

return [{
    "json": {
        "total": total,
        "count": len(valid_items)
    }
}]

3. 使用正则表达式处理字符串

从文本中提取模式

import re

items = _input.all()
email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'

all_emails = []
for item in items:
    text = item["json"].get("text", "")
    emails = re.findall(email_pattern, text)
    all_emails.extend(emails)

# Remove duplicates
unique_emails = list(set(all_emails))

return [{
    "json": {
        "emails": unique_emails,
        "count": len(unique_emails)
    }
}]

4. 数据验证

验证并清理数据

items = _input.all()
validated = []

for item in items:
    data = item["json"]
    errors = []

    # Validate fields
    if not data.get("email"):
        errors.append("Email required")
    if not data.get("name"):
        errors.append("Name required")

    validated.append({
        "json": {
            **data,
            "valid": len(errors) == 0,
            "errors": errors if errors else None
        }
    })

return validated

5. 统计分析

使用 statistics 模块计算统计数据

from statistics import mean, median, stdev

items = _input.all()
values = [item["json"].get("value", 0) for item in items if "value" in item["json"]]

if values:
    return [{
        "json": {
            "mean": mean(values),
            "median": median(values),
            "stdev": stdev(values) if len(values) > 1 else 0,
            "min": min(values),
            "max": max(values),
            "count": len(values)
        }
    }]
else:
    return [{"json": {"error": "No values found"}}]

参见：COMMON_PATTERNS.md，了解 10 种详细的 Python 模式

---

错误预防——最常见的 5 个错误

#1：导入外部库（Python 专属！）

# ❌ WRONG: Trying to import external library
import requests  # ModuleNotFoundError!

# ✅ CORRECT: Use HTTP Request node or JavaScript
# Add HTTP Request node before Code node
# OR switch to JavaScript and use $helpers.httpRequest()

#2：代码为空或缺少返回值

# ❌ WRONG: No return statement
items = _input.all()
# Processing...
# Forgot to return!

# ✅ CORRECT: Always return data
items = _input.all()
# Processing...
return [{"json": item["json"]} for item in items]

#3：返回格式不正确

# ❌ WRONG: Returning dict instead of list
return {"json": {"result": "success"}}

# ✅ CORRECT: List wrapper required
return [{"json": {"result": "success"}}]

#4：访问字典时出现 KeyError

# ❌ WRONG: Direct access crashes if missing
name = _json["user"]["name"]  # KeyError!

# ✅ CORRECT: Use .get() for safe access
name = _json.get("user", {}).get("name", "Unknown")

#5：Webhook 请求体嵌套

# ❌ WRONG: Direct access to webhook data
email = _json["email"]  # KeyError!

# ✅ CORRECT: Webhook data under ["body"]
email = _json["body"]["email"]

# ✅ BETTER: Safe access with .get()
email = _json.get("body", {}).get("email", "no-email")

参见：ERROR_PATTERNS.md，了解全面的错误指南

---

标准库参考

最实用的模块

# JSON operations
import json
data = json.loads(json_string)
json_output = json.dumps({"key": "value"})

# Date/time
from datetime import datetime, timedelta
now = datetime.now()
tomorrow = now + timedelta(days=1)
formatted = now.strftime("%Y-%m-%d")

# Regular expressions
import re
matches = re.findall(r'\d+', text)
cleaned = re.sub(r'[^\w\s]', '', text)

# Base64 encoding
import base64
encoded = base64.b64encode(data).decode()
decoded = base64.b64decode(encoded)

# Hashing
import hashlib
hash_value = hashlib.sha256(text.encode()).hexdigest()

# URL parsing
import urllib.parse
params = urllib.parse.urlencode({"key": "value"})
parsed = urllib.parse.urlparse(url)

# Statistics
from statistics import mean, median, stdev
average = mean([1, 2, 3, 4, 5])

参见：STANDARD_LIBRARY.md，了解完整参考

---

最佳实践

1. 访问字典时始终使用 .get()

# ✅ SAFE: Won't crash if field missing
value = item["json"].get("field", "default")

# ❌ RISKY: Crashes if field doesn't exist
value = item["json"]["field"]

2. 明确处理 None/Null 值

# ✅ GOOD: Default to 0 if None
amount = item["json"].get("amount") or 0

# ✅ GOOD: Check for None explicitly
text = item["json"].get("text")
if text is None:
    text = ""

3. 使用列表推导式进行筛选

# ✅ PYTHONIC: List comprehension
valid = [item for item in items if item["json"].get("active")]

# ❌ VERBOSE: Manual loop
valid = []
for item in items:
    if item["json"].get("active"):
        valid.append(item)

4. 返回一致的结构

# ✅ CONSISTENT: Always list with "json" key
return [{"json": result}]  # Single result
return results  # Multiple results (already formatted)
return []  # No results

5. 使用 print() 语句进行调试

# Debug statements appear in browser console (F12)
items = _input.all()
print(f"Processing {len(items)} items")
print(f"First item: {items[0] if items else 'None'}")

---

何时使用 Python，何时使用 JavaScript

使用 Python 的情况：

• ✅ 需要使用 statistics 模块进行统计运算
• ✅ 你对 Python 语法明显更熟悉
• ✅ 你的逻辑适合用列表推导式表示
• ✅ 需要特定的标准库函数

使用 JavaScript 的情况：

• ✅ 需要发送 HTTP 请求（$helpers.httpRequest()）
• ✅ 需要高级日期/时间功能（DateTime/Luxon）
• ✅ 希望更好地集成 n8n
• ✅ 适用于 95% 的使用场景（推荐）

考虑使用其他节点的情况：

• ❌ 简单的字段映射 → 使用 Set 节点
• ❌ 基本筛选 → 使用 Filter 节点
• ❌ 简单条件判断 → 使用 IF 或 Switch 节点
• ❌ 仅需发送 HTTP 请求 → 使用 HTTP Request 节点

---

与其他技能集成

可搭配使用：

n8n 表达式语法：

• 其他节点中的表达式使用 {{ }} 语法
• Code 节点直接使用 Python（无需 {{ }}）
• 了解何时使用表达式，何时使用代码

n8n MCP 工具专家：

• 查找 Code 节点：search_nodes({query: "code"})
• 获取配置帮助：get_node_essentials("nodes-base.code")
• 验证代码：validate_node_operation()

n8n 节点配置：

• 模式选择（All Items 与 Each Item）
• 语言选择（Python 与 JavaScript）
• 了解属性依赖关系

n8n 工作流模式：

• 在转换步骤中使用 Code 节点
• 了解在不同模式中何时使用 Python，何时使用 JavaScript

n8n 验证专家：

• 验证 Code 节点配置
• 处理验证错误
• 自动修复常见问题

n8n Code JavaScript：

• 了解何时改用 JavaScript
• 比较 JavaScript 与 Python 的功能
• 从 Python 迁移到 JavaScript

---

快速参考清单

部署 Python 代码节点之前，请确认：

• 优先考虑 JavaScript - 仅在必要时使用 Python
• 代码不为空 - 必须包含有意义的逻辑
• 存在返回语句 - 必须返回字典列表
• 返回格式正确 - 每个项目：{"json": {...}}
• 数据访问正确 - 使用 _input.all()、_input.first() 或 _input.item
• 不导入外部模块 - 仅使用标准库（json、datetime、re 等）
• 安全访问字典 - 使用 .get() 以避免 KeyError
• Webhook 数据 - 如果数据来自 Webhook，通过 ["body"] 访问
• 模式选择 - 大多数情况下选择“All Items”
• 输出一致 - 所有代码路径都返回相同结构

---

其他资源

相关文件

• DATA_ACCESS.md - Python 数据访问模式详解
• COMMON_PATTERNS.md - 适用于 n8n 的 10 种 Python 模式
• ERROR_PATTERNS.md - 常见的 5 种错误及解决方案
• STANDARD_LIBRARY.md - 完整的标准库参考

n8n 文档

• Code Node 指南：https://docs.n8n.io/code/code-node/
• n8n 中的 Python：https://docs.n8n.io/code/builtin/python-modules/

---

准备好在 n8n Code 节点中编写 Python 了吗？但请优先考虑 JavaScript！ 根据具体需求使用 Python，参考错误模式指南以避免常见错误，并有效利用标准库。