Open API 和 SDK

于 2.6.4 版本新增

DataFlux Func 提供了完整的 Open API 支持，可以使用配套的 DataFlux Func SDK 通过编程方式调用。

需要直接在本地 Python 环境运行脚本时，可以使用本地脚本引擎，无需创建 Access Key 或连接 DataFlux Func 服务。

1. 开启 Open API 文档页面

在「实验性功能」中，可以开启 Open API 的文档页。

DataFlux Func SDK 包含了签名功能，且以单文件方式发布。用户可以直接放入项目中使用。

2. 创建 Access Key

登录你的 DataFlux Func
在「管理 / 实验性功能」中启用 Access Key 管理
在「管理 / Access Key」点击「新建」创建 Access Key

3. 使用 SDK 发送请求

DataFlux Func SDK 支持多种编程语言，下载地址如下：

语言	下载地址	说明
Python	dataflux_func_sdk.py	单文件，基础使用无依赖。上传功能依赖 `requests` 库
Node.js	dataflux_func_sdk.js	单文件，基础使用无依赖。上传功能依赖 `form-data` 库
Golang	dataflux_func_sdk.go	单文件，完全无依赖 Demo: dataflux_func_sdk_demo.go

发送请求示例如下：

PythonNode.jsGolang

Python
from dataflux_func_sdk import DataFluxFunc

# 创建 DataFlux Func 操作对象
dff = DataFluxFunc(ak_id='ak-xxxxx', ak_secret='xxxxxxxxxx', host='localhost:8088')

# 开启 Debug
dff.debug = True

# 发送 GET 请求
try:
    status_code, resp = dff.get('/api/v1/do/ping')

except Exception as e:
    print(colored(e, 'red'))
    raise

# 发送 POST 请求
try:
    body = {
        'echo': {
            'int'    : 1,
            'str'    : 'Hello World',
            'none'   : None,
            'boolean': True,
        }
    }
    status_code, resp = dff.post('/api/v1/do/echo', body=body)

except Exception as e:
    print(colored(e, 'red'))
    raise

# 上传文件
try:
    filename = 'your_file'
    with open(filename, 'rb') as _f:
        file_buffer = _f.read()

    fields = {
        'folder': 'test'
    }

    status_code, resp = dff.upload('/api/v1/resources/do/upload',
        file_buffer=file_buffer,
        filename=filename,
        fields=fields)

except Exception as e:
    print(colored(e, 'red'))
    raise

JavaScript
var fs = require('fs');
var DataFluxFunc = require('./dataflux_func_sdk.js').DataFluxFunc;

// 创建 DataFlux Func 操作对象
var opt = {
  akId    : 'ak-xxxxx',
  akSecret: 'xxxxxxxxxx',
  host    : 'localhost:8088',
};
var dff = new DataFluxFunc(opt);

// 开启 Debug
dff.debug = true;

// 发送 GET 请求
var getOpt = {
  path: '/api/v1/do/ping',
};
dff.get(getOpt, function(err, respData, respStatusCode) {
  if (err) console.error(colored(err, 'red'))

  // 发送 POST 请求
  var postOpt = {
    path: '/api/v1/do/echo',
    body: {
      'echo': {
        'int'    : 1,
        'str'    : 'Hello World',
        'none'   : null,
        'boolean': true,
      }
    }
  };
  dff.post(postOpt, function(err, respData, respStatusCode) {
    if (err) console.error(colored(err, 'red'))

    // 上传文件
    var filename = 'your_file';
    var uploadOpt = {
      path      : '/api/v1/resources/do/upload',
      fileBuffer: fs.readFileSync(filename),
      filename  : filename,
      fields    : {
        'folder': 'test'
      },
    };
    dff.upload(uploadOpt, function(err, respData, respStatusCode) {
      if (err) console.error(colored(err, 'red'))
    });
  });
});

Go
package main

import (
    "os"
    "bytes"
    "fmt"
    "io"
    "mime/multipart"

    // DataFlux Func SDK
    "./dataflux_func_sdk"
)

var (
    colorMap = map[interface{}]string{
        "grey":    "\033[0;30m",
        "red":     "\033[0;31m",
        "green":   "\033[0;32m",
        "yellow":  "\033[0;33m",
        "blue":    "\033[0;34m",
        "magenta": "\033[0;35m",
        "cyan":    "\033[0;36m",
    }
)

func main() {
    host := "localhost:8088"
    if len(os.Args) >= 2 {
        host = os.Args[1]
    }

    // 创建 DataFlux Func 操作对象
    dff := dataflux_func_sdk.NewDataFluxFunc("ak-xxxxx", "xxxxxxxxxx", host, 30, false)

    // 开启 Debug
    dff.Debug = true

    // 发送 GET 请求
    _, _, err := dff.Get("/api/v1/do/ping", nil, nil, "")
    if err != nil {
        panic(err)
    }

    // 发送 POST 请求
    body := map[string]interface{}{
        "echo": map[string]interface{}{
            "int"    : 1,
            "str"    : "Hello World",
            "none"   : nil,
            "boolean": true,
        },
    }
    _, _, err = dff.Post("/api/v1/do/echo", body, nil, nil, "")
    if err != nil {
        panic(err)
    }

    // 上传文件
    filename := "dataflux_func_sdk_demo.go"
    file, _ := os.Open(filename)
    fileContents, _ := io.ReadAll(file)

    uploadBody := &bytes.Buffer{}
    writer := multipart.NewWriter(uploadBody)

    part, _ := writer.CreateFormFile("files", filename)
    part.Write(fileContents)

    fields := map[string]string{
        "folder": "test",
    }
    for key, value := range fields {
        writer.WriteField(key, fmt.Sprintf("%v", value))
    }

    writer.Close()
    contentType := writer.FormDataContentType()

    _, _, err = dff.Upload("/api/v1/resources/do/upload", filename, "", fields, nil, nil, uploadBody, contentType)
    if err != nil {
        panic(err)
    }
}

4. 本地执行脚本

于 8.1.15 版本新增

从 DataFlux Func 8.1.15 或更高版本源码中取得 sdk/dataflux_func_local.py，放入本地 Python 项目即可使用。本地引擎是仅依赖 Python 标准库的单文件工具，要求 Python 3.8 或更高版本，可独立于 HTTP SDK、Server、Worker、数据库和 Redis 运行。

脚本自身使用的第三方 Python 包需要安装到同一 Python 环境中，引擎不会自动安装依赖。

4.1 执行函数

将引擎和脚本放在同一目录，例如：

Text Only
my-scripts/
  dataflux_func_local.py
  demo__example.py

demo__example.py 内容如下：

Python
@DFF.API('计算两数之和')
def plus(x, y):
    return x + y

在该目录执行：

Bash
python dataflux_func_local.py run demo__example.plus --kwargs '{"x": 1, "y": 2}'

标准输出为 3。--kwargs 必须是 JSON 对象，省略时使用 {}；也可以调用未使用 @DFF.API 修饰的普通函数。

4.2 脚本目录与导入

脚本根目录默认是启动命令时的当前工作目录，可通过 --root 指定：

Bash
python dataflux_func_local.py run demo__example.plus --root ./exported --kwargs '{"x": 1, "y": 2}'

对于 demo__example.plus，支持以下相对于根目录的脚本路径：

Text Only
demo__example.py
demo__example.draft.py
script-sets/demo/example.py
script-sets/demo/demo__example.py

--root 也可以直接指向 script-sets 目录。每个脚本 ID 只保留一个匹配文件；发现多个匹配文件时会报告歧义。引擎读取本地文件内容，不区分草稿和已发布版本。

脚本可以用 import __helper 导入同一脚本集的脚本，也可以用 import other__helper 导入其他脚本集的脚本。运行期间每个导入脚本只加载一次并复用其模块。所需脚本应全部放在指定根目录内；引擎不解析 META.yaml，也不会自动下载缺失的脚本或资源。

4.3 支持的脚本能力

功能	本地行为
`DFF.API`	记录函数元数据和常用装饰器参数，由命令显式调用指定函数
`print`、`DFF.LOG`、`DFF.VAR`	日志写入标准错误；Python、原生库及继承标准输出的子进程输出也会重定向到标准错误
`DFF.ENV`	从进程环境读取字符串，支持直接调用及 `get`、`keys`、`ref`；不提供平台环境变量类型和自动密码遮蔽
`DFF.CTX`	保存本次运行的上下文；读取时通过 JSON 复制快照，保存的值应兼容 JSON
`DFF.RSRC`	将资源相对路径解析到 `<root>/resources` 下，供 Python 文件操作使用，路径不能越出该目录
`DFF.TEMP_DIR`	首次使用时创建临时目录，可主动清空，执行结束或抛出异常后自动清理
`DFF.RESP`	输出包含 `returnValue` 和 `responseControl` 的 JSON 对象，不启动 HTTP 服务

本地执行使用调试模式，_DFF_IS_DEBUG、_DFF_DEBUG 为 True，_DFF_IS_PUBLISH 为 False；HTTP 请求和定时任务上下文为空。

DFF.API 中的超时、缓存、队列、定时任务、HTTP API、认证和自动运行等参数仅作为声明保留，不会执行对应的平台行为；入口函数声明了这些参数时，引擎会向标准错误输出提示。特别是，达到声明的超时时间后，引擎不会中止函数。

自定义函数元数据可使用 custom 或 custom_json；custom_yaml 不受支持。连接器、持久存储、简易缓存、线程池和平台管理等未列出的能力也不受支持，访问时会明确报错。

4.4 结果与错误

执行成功时，标准输出包含一个 JSON 结果；函数返回值应可序列化为 JSON，NaN 和 Infinity 会被拒绝。日志和错误写入标准错误，因此可以单独保存结果：

Bash
python dataflux_func_local.py run demo__example.plus --kwargs '{"x": 1, "y": 2}' > result.json

参数错误（包括无效的 --kwargs）退出码为 2；脚本、导入或结果序列化错误退出码为 1，异常堆栈指向本地源文件；用户中断时退出码为 130。