Python 入门 / 代码片段参考

本文汇总 DataFlux Func 脚本中常见的 Python 处理方式。示例可以作为起点，但应根据真实数据结构、失败策略和性能要求进行调整。

1. 转换列表中的数据

逻辑简单时，可以使用列表推导式生成新列表。以下示例将每个数据点的值平方，同时保留时间戳：

Python
dps = [
    [1583172563000, 0],
    [1583172564000, 1],
    [1583172565000, 2],
    [1583172566000, 3],
    [1583172567000, 4],
]

result = [[timestamp, value ** 2] for timestamp, value in dps]
print(result)

输出：

Text Output
[[1583172563000, 0], [1583172564000, 1], [1583172565000, 4], [1583172566000, 9], [1583172567000, 16]]

需要校验、跳过异常值或执行多步处理时，普通 for 循环更清晰：

Python
dps = [
    [1583172563000, 2],
    [1583172564000, None],
    [1583172565000, 4],
]

result = []
for timestamp, value in dps:
    if value is None:
        continue

    result.append([timestamp, value ** 2])

print(result)

输出：

Text Output
[[1583172563000, 4], [1583172565000, 16]]

不要为了缩短行数把复杂分支全部塞进推导式；可读性比少写几行更重要。

2. 读取嵌套字典中的值

数据结构固定且缺少任意一层都视为“没有值”时，可以逐层使用 dict.get()：

Python
input_data = {
    'level1': {
        'level2': {
            'level3': 'value',
        },
    },
}

level1 = input_data.get('level1', {})
level2 = level1.get('level2', {})
value  = level2.get('level3')

print(value)

输出：

Text Output
value

如果缺少键表示输入格式错误，应使用索引读取，并只捕获预期异常：

Python
input_data = {'level1': {}}

try:
    value = input_data['level1']['level2']['level3']
except (KeyError, TypeError) as e:
    print(f'输入数据结构不符合要求：{e}')

不要使用 except Exception 后直接 pass，否则拼写错误、类型错误等程序缺陷也会被静默隐藏。

3. 去重、筛选与排序

3.1 保留顺序去重

对可哈希元素组成的列表，可以使用字典键去重并保留首次出现的顺序：

Python
items        = ['warning', 'error', 'warning', 'info', 'error']
unique_items = list(dict.fromkeys(items))

print(unique_items)

输出：

Text Output
['warning', 'error', 'info']

直接使用 set(items) 也能去重，但结果没有固定顺序。

3.2 筛选并排序字典列表

Python
events = [
    {'name': 'cpu', 'value': 72},
    {'name': 'memory', 'value': 91},
    {'name': 'disk', 'value': 85},
]

alerts = [event for event in events if event['value'] >= 80]
alerts = sorted(alerts, key=lambda event: event['value'], reverse=True)

print(alerts)

输出：

Text Output
[{'name': 'memory', 'value': 91}, {'name': 'disk', 'value': 85}]

4. 生成带时区的时间范围

查询接口经常需要明确的开始时间和结束时间。使用 arrow 可以生成最近一小时的 UTC 时间范围：

Python
import arrow

end_time   = arrow.utcnow()
start_time = end_time.shift(hours=-1)

params = {
    'start': start_time.isoformat(),
    'end'  : end_time.isoformat(),
}

print(params)

实际输出取决于运行时刻，但 start 和 end 都是带 UTC 偏移的 ISO 8601 字符串，且相差一小时。

如果接口要求毫秒时间戳，可以明确转换单位：

Python
import arrow

current                = arrow.get('2026-08-11T10:30:00+08:00')
timestamp_milliseconds = current.int_timestamp * 1000

print(timestamp_milliseconds)

输出：

Text Output
1786415400000

int_timestamp 返回整数秒，乘以 1000 后得到整数毫秒。不要同时把秒级和毫秒级时间戳传给同一接口。

5. 发送 HTTP 请求

发送请求时至少应设置超时、检查状态码，并只解析预期格式的响应：

Python
import requests

params = {
    'status': 'active',
}

try:
    response = requests.get('https://httpbin.org/get', params=params, timeout=10)
    response.raise_for_status()
    data = response.json()
except requests.JSONDecodeError as e:
    print(f'响应不是有效的 JSON：{e}')
except requests.RequestException as e:
    print(f'请求失败：{e}')
else:
    print(data['args'])

请求失败后的策略取决于业务：可以有限重试、返回明确错误或让异常继续抛出。不要无限重试，也不要在日志中输出访问令牌、完整请求头或敏感响应体。

6. 发送 Webhook 消息

多数 Webhook 接口接收 JSON 请求体。以下为通用结构，字段需要按目标平台的协议调整：

Python
import requests

webhook_url = 'https://example.com/webhook'
payload     = {
    'type': 'text',
    'text': {
        'content': 'DataFlux Func 告警测试',
    },
}

response = requests.post(webhook_url, json=payload, timeout=10)
response.raise_for_status()

真实平台可能还要求签名、时间戳、关键词或来源 IP 校验。Webhook 地址通常包含凭证，应从安全配置读取，不能提交到代码仓库或写入普通日志。

7. 使用参数化 SQL

把用户输入直接拼接到 SQL 字符串中会产生 SQL 注入风险。DataFlux Func 的 SQL 连接器支持参数化查询，应把值放入 sql_params：

Python
target_id = 'user-001'

sql        = 'SELECT * FROM users WHERE id = ?'
sql_params = [target_id]
rows       = helper.query(sql, sql_params=sql_params)

不要使用以下写法：

Python
sql  = f"SELECT * FROM users WHERE id = '{target_id}'"
rows = helper.query(sql)

表名和字段名不能作为普通值参数。确实需要动态标识符时，只能从可信配置或允许列表中选择，再按连接器支持的方式传入；不要接受任意用户输入。

不同连接器的查询方法和返回结构略有差异，请参考脚本开发 / 连接器对象 DFF.CONN / 总览和脚本开发 / SQL 构造 DFF.SQL。