search
DataFlux.f(x) 脚本编写

脚本编写

DataFlux.f(x)使用Python 3.7.5 作为运行环境,并对脚本的执行上下文进行了一些定制,与原版的Python有一定区别。 根据DataFlux.f(x)要求,本系统中的Python脚本都应该使用4个空格作为缩进单位,请勿使用tab等其他缩进方式。 此外,由于脚本中的函数返回值最终都会通过HTTP协议,以JSON数据返回给调用方。因此,函数返回值要求能够被序列化为JSON数据,一些复杂对象实例无法直接返回。

最简单的函数

一个最简单的函数如下:

def hello_world():
    return 'Hello, World!'

可以看到,这个函数和原版Python中的函数写法并没有任何区别。 大部分情况下,在DataFlux.f(x)中编写函数也与和原版Python没有什么区别。

此时,hello_world()函数已经可以使用,但仅限DataFlux.f(x)内部被其他函数调用。如果希望函数能够在DataFlux.f(x)外部被调用,则需要添加 @DFF.API() 装饰器。

装饰器的参数列表如下:

参数 是否必须 说明
title 必须 被修饰函数的中文标题。主要用于函数文档、UI界面的展示,本身并没有具体功能
category 可选,默认None 被修饰函数的分类,主要用于与DataFlux Studio对接等
tags 可选,默认None 被修饰函数的标签,字符串数组,主要用于函数文档API的查询过滤
fixed_crontab 可选,默认None 被修饰函数以自动触发方式运行时,以此固定的crontab表达式运行

典型示例如下:

@DFF.API('一个最简单的函数', category='action', tags=['hello', 'world'], fixed_crontab='*/15 * * * *')
def hello_world():
    return 'Hello, World!'

除了直接返回字符串,也可以返回能够序列化为JSON的对象。函数本身不需要手工进行序列化操作,如:

@DFF.API('一个最简单的函数')
def hello_world():
    ret = { 'message': 'Hello, World!' }
    return ret

如果确实没有返回值需要返回,返回None即可(Python没有return语句时,同样返回None),如:

@DFF.API('一个没有返回值的函数')
def hello_world():
    return None

函数可以附带参数、函数文档,同样按照正常Python编写即可,如:

@DFF.API('问候')
def hello_world(you_name=None):
    '''
    本函数返回一句问候语
    参数:
        your_name 可选,字符串。问候者姓名
    返回:
        字符串
    '''
    if you_name is None:
        your_name = 'My friend'
    return 'Hello, World! ' + your_name

在发布脚本后,DataFlux.f(x)会提取函数的标题、名称、参数列表、函数文档等,供DataFlux使用或通过函数文档对外展示。因此请务必正确填写。

编写函数的要求

在DataFlux.f(x)中编写函数,存在一些限制和要求。有些是出于安全考虑,有些是为了保证和DataFlux能够正常交互,还有一些主要是考虑到用户实际操作的便利性。

在编写函数时,请务必遵守这些规定,否则可能会带来意想不到的问题

允许import的Python内置库

出于安全性方面的考虑,在DataFlux.f(x)中只允许import部分内置库:

  • base64
  • collections
  • datetime
  • dns
  • hashlib
  • hmac
  • itertools
  • json
  • math
  • random
  • re
  • time
  • uuid
  • pprint
  • functools
  • gzip

允许import的第三方Python库

除了Python的内置库,DataFlux.f(x)预装了部分常用的第三方库:

  • six (v1.12.0)

  • arrow (v0.13.0)

  • requests (v2.21.0)

  • simplejson (v3.16.0)

  • ujson (v1.35)

  • xmltodict (v0.9.2)

  • PyYAML (v4.2b2)

  • numpy (v1.18.1)

  • pandas (v0.25.3)

  • scipy (v1.3.3)

  • scikit-learn (v0.22.1)

  • statsmodels (v0.11.0)

  • fbprophet (v0.5)

  • eas_prediction (v0.12)

函数名称

函数名称即def后面的函数名称,在DataFlux.f(x)中也被称为引用名。 除了需要满足Python本身的要求外,在DataFlux.f(x)中额外要求长度不允许超过256个字符。

函数定义

由于代码解析上的限制,函数定义无论多长都必须写在一行里,如:

def hello_world(arg1=None, arg2=None, arg3=None, args4=None):
    pass

以下为错误示范:

def hello_world(arg1=None, arg2=None,
    arg3=None, args4=None):
    pass

函数标题

函数标题为 @DFF.API()装饰器的第一个参数title,在DataFlux.f(x)中要求长度不超过256个字符。

函数文档

函数文档为函数定义行(def行)之后,第一个代码块,如:

@DFF.API('问候')
def hello_world(you_name=None):
    '''
    本函数返回一句问候语
    参数:
        your_name 可选,字符串。问候者姓名
    返回:
        字符串
    '''
    if you_name is None:
        your_name = 'My friend'
    return 'Hello, World! ' + your_name

编写时,缩进请勿小于三引号,并保证三引号都各自单独占有一行。

以下为错误示范:

def hello_world():
    '''这是函数文档的错误示范'''
    pass

def hello_world():
    '''
    这是函数文档的错误示范'''
    pass

def hello_world():
    '''这是函数文档的错误示范
    '''
    pass

def hello_world():
    '''
这是函数文档的错误示范
    '''
    pass

如内容比较多,需要分节展示,可以使用悬挂缩进方式:

def hello_world():
    '''
    内容比较多需要分节时,可以悬挂缩进
    第一节标题
        第一节内容
    第二节标题
        第二节内容
    '''
    pass

标准模板:

def hello_world(you_name=None):
    '''
    本函数返回一句问候语
    参数:
        your_name 可选,字符串。问候者姓名
    返回:
        字符串
    '''
    pass

函数参数

函数参数本质上与原版Python并没有什么区别,但是由于DataFlux.f(x)中暴露的函数需要与DataFlux Studio进行交互,必须考虑用户在Web页面上操作的便利性。因此,对于被@DFF.API()修饰的函数,需要按照一定规范设计参数列表。

不要使用Python参数解包功能(\args, **kwargs)*

虽然Python参数解包功能非常便利,但这样的参数无法在Web页面供用户填写参数,也无法进行参数提示,因此要避免这种设计。

@DFF.API('错误示范')
def hello_world(*args, **kwargs):
    pass

@DFF.API('正确示范')
def hello_world(message=None):
    pass

函数返回值

函数返回值本质上与原版Python并没有什么区别,但是由于DataFlux.f(x)中暴露的函数需要与DataFlux Studio进行交互,必须考虑返回值可以DataFlux Studio正常解析。因此,对于被@DFF.API()修饰的函数,需要按照一定规范设计返回值。

总是返回能够被序列化为JSON的数据

在Python,可以被正常序列化的数据为dict、list、tuple、int、float、str、unicode、bool、None等基本内置数据类型。

此外,对于第三方库特有的对象实例,如果实现了对应的序列化方法(即虽然不是内置数据类型,但依然可以序列化为JSON)也可以返回。

@DFF.API('正确示范')
def hello_world():
    '''
    返回可序列化为JSON的数据
    '''
    ret = {
        'list': [1, 1.1, True, False, None],
        'dict': {'int': 1, 'float': 1.1, 'bool': True, 'null': None},
        'int': 1,
        'float': 1.1,
        'bool': True,
        'null': None,
    }

不要返回多个返回值

由于函数返回值在JSON序列化后会包裹在result字段中,方便程序获取,而返回多个返回值会导致result变成数组。

@DFF.API('错误示范')
def hello_world():
    '''
    请勿返回多个返回值
    '''
    return True, [1, 2, 3]

@DFF.API('正确示范')
def hello_world():
    '''
    一次只返回一个返回值
    '''
    ret = {
        'ok': True,
        'data': [1, 2, 3]
    }
    return ret

错误应当使用raise语句抛出,而不是return

由于DataFlux.f(x)本身需要监控函数执行成功与否。因此,当函数执行发生错误时,应当使用raise语句将问题抛出,而不是return。否则DataFlux.f(x)无法获知函数是否正确执行。

@DFF.API('错误示范')
def hello_world():
    '''
    请勿使用return代替raise
    '''
    return False

@DFF.API('正确示范')
def hello_world():
    '''
    出错应当使用raise语句
    '''
    raise Exception('Some error message')

使用内置功能

为了方便脚本编写,以及在脚本中使用各种 DataFlux.f(x)提供的功能,DataFlux.f(x)在脚本运行上下文注入了一些额外功能。使用这些功能不需要额外 import,直接使用即可。这些功能都封装在 DFF 对象中(如上文出现的@DFF.API)

将函数对外暴露(DFF.API)

一个装饰器,用于将被修饰的函数对外暴露,允许使用API方式调用。如:

@DFF.API('问候')
def hello_world(you_name=None):
    '''
    本函数返回一句问候语
    参数:
        your_name 可选,字符串。问候者姓名
    返回:
        字符串
    '''
    if you_name is None:
        your_name = 'My friend'
    return 'Hello, World! ' + your_name

此外,DFF.API还可以额外指定分类category,标示特定用途,如:

@DFF.API('一个预测函数', category='prediction')
def demo_prediction(dps):
    '''
    一个预测函数
    '''
    pass

有关分类category的内容,详见下文「附带分类的函数」

操作数据源

在脚本编辑器左侧边栏配置的、内置的所有数据源,都可以在脚本中使用配置的「引用名」获取对应的数据源操作对象。需要查询数据时,使用此操作对象即可,如:

@DFF.API('InfluxDB操作演示')
def influxdb_demo():
    '''
    本函数从数据源`demo_influxdb`中查询3条`demo`指标并返回
    '''
    db = DFF.SRC('demo_influxdb')
    return db.query('SELECT * FROM demo LIMIT 3')

如数据源配置了默认数据库,则查询会在默认数据库进行。

如数据源没有配置默认数据库,或在查询时需要查询不同的数据库,可在获取数据源操作对象时,指定数据库database参数,如:

db = DFF.SRC('demo_influxdb', database='my_database')

对于DataFlux DataWay,可以获取数据源操作对象时指定DataWay的rp和token参数,如:

dataway = DFF.SRC('df_dataway', token='xxxxx', rp='rp0')

由于数据源具有不同类型,各个数据源操作对象略有区别。 详细请参考《DataFlux.f(x) 包学包会》

获取环境变量

在脚本编辑器左侧边栏配置的所有环境变量,都可以在脚本中使用配置的「引用名」获取对应的环境变量值。

示例代码如下:

company_name = DFF.ENV('companyName')
# '上海驻云信息科技有限公司'

由于环境变量都是在Web中进行配置,因此所有的环境变量值都是字符串类型。使用时需要按照需要进行类型转换,如:

try:
    page_size = int(DFF.ENV('pageSize'))
except Exception as e:
    pass

输出日志

由于脚本编辑器是Web应用程序,不同于一般意义上的IDE,无法进行单步调试等操作。因此,调试主要依靠运行时输出日志进行。

为了让日志输出能够被DataFlux.f(x)搜集并显示到脚本编辑器中,DataFlux.f(x)提供了专门的DFF.log()日志输出函数。

输出日志

附带分类的函数

在使用 @DFF.API()装饰器时,可以额外指定分类category参数。category参数是字符串枚举值,包含以下选项:

  • prediction:预测函数
  • transformation:转换函数
  • action:动作函数
  • command:命令函数
  • query:查询函数
  • check:检测函数

指定了分类参数的函数,会被DataFlux获取,并可以直接在DataFlux Studio特定的页面中使用。因此不同的分类函数具有额外的设计规范。

DataFlux标准数据结构

由于需要与DataFlux Studio进行交互,DataFlux规定了一些标准的数据结构。这些结构在附带分类的函数中非常常见,且作为参数时,参数名必须固定为其英文名

注意:函数的其他参数不要与固定参数名相同。

dps时序数据

dps时序数据为DataFlux Studio图表中的系列数据。一个dps时序数据在DataFlux Studio的图表中为一个数据系列,如一条折线、一排柱状图等。

dps时序数据的固定参数名为dps,结构描述如下:

[[UNIX毫秒时间戳,值], [UNIX毫秒时间戳,值], ...]

具体示例如下:

[
  [ 1577808000000, 100 ],
  [ 1577808060000, 120 ],
  ...
]

start_time、end_time起止时间

start_time、end_time起止时间一般为图表操作时指定的时间范围端点,为UNIX毫秒时间戳。可能由用户选择输入,也有可能为DataFlux Studio全局时间范围填入。

start_time、end_time起止时间的固定参数名为start_time、end_time。

具体示例如下:

1577808000000

预测函数

预测函数主要用于DataFlux Studio中图表分析模式,接收图表中的dps时序数据,返回预测后的dps时序数据。规范如下:

  • 分类category为prediction。
  • 第1个参数为固定参数,必须为dps时序数据。值为图表中单个数据系列的时序数据
  • 可以不附带或附带一个或多个可选参数。
  • 返回值同样为dps时序数据。内容为预测所得的时序数据

示例如下:

@DFF.API('预测函数1', category='prediction')
def my_prediction_1(dps):
    '''
    一个预测函数,只需要固定参数dps
    '''
    # 进行数据预测,结果保存在result变量中
    return result

@DFF.API('预测函数2', category='prediction')
def my_prediction_2(dps, param1=None, param2=None):
    '''
    一个预测函数,需要固定参数dps以及额外参数
    '''
    # 进行数据预测,结果保存在result变量中
    return result

转换函数

转换函数主要用于DataFlux Studio中图表分析模式,接收图表中的dps时序数据,返回转换后的dps时序数据,规范如下:

  • 分类category为transformation。
  • 第1个参数为固定参数,必须为dps时序数据。值为图表中单个数据系列的时序数据
  • 可以不附带或附带一个或多个可选参数。
  • 返回值同样为dps时序数据。内容为转换所得的时序数据   示例如下:
@DFF.API('转换函数1', category='transformation')
def my_transformation_1(dps):
    '''
    一个转换函数,只需要固定参数dps
    '''
    # 进行数据转换,结果保存在result变量中
    return result

@DFF.API('转换函数2', category='transformation')
def my_transformation_2(dps, param1=None, param2=None):
    '''
    一个转换函数,需要固定参数dps以及额外参数
    '''
    # 进行数据转换,结果保存在result变量中
    return result

行为函数

行为函数主要用于DataFlux Studio中基线告警的触发执行。没有特定的参数和返回值要求,规范如下:

  • 分类category为action
  • 可以不附带或附带一个或多个可选参数
  • 不需要返回值

示例如下:

@DFF.API('行为函数1', category='action')
def my_action_1():
    '''
    一个行为函数,没有参数和返回值要求
    '''
    pass

@DFF.API('行为函数2', category='action')
def my_action_2(param1=None, param2=None):
    '''
    一个行为函数,没有参数和返回值要求
    '''
    pass

注意:「行为函数」有时候会被称为「动作函数」,是相同意思

命令函数

命令函数主要用于DataFlux Studio中的点击自定义按钮的触发执行。没有特定的参数和返回值要求,规范如下:

  • 分类category为command
  • 可以不附带或附带一个或多个可选参数
  • 不需要返回值

示例如下:

@DFF.API('命令函数1', category='command')
def my_command_1():
    '''
    一个命令函数,没有参数和返回值要求
    '''
    pass

@DFF.API('命令函数2', category='command')
def my_command_2(param1=None, param2=None):
    '''
    一个命令函数,没有参数和返回值要求
    '''
    pass

查询函数

查询函数主要用于DataFlux Studio中自定义查询展示图表的功能,可以接收图表界面指定的start_time、end_time起止时间(也可以不接收),返回查询获得的dps时序数据,规范如下:

  • 分类category为query
  • 如存在参数start_time、end_time起止时间,则必须为第1、第2个参数,值为UNIX毫秒时间戳
  • 可以不附带或附带一个或多个可选参数
  • 返回值为dps时序数据。内容为查询所得的时序数据   示例如下:
@DFF.API('查询函数1', category='query')
def my_query_1():
    '''
    一个查询函数,没有参数
    '''
    # 执行查询,结果保存在result中
    return result

@DFF.API('查询函数2', category='query')
def my_query_2(start_time, end_time):
    '''
    一个查询函数,包含起止时间参数
    '''
    # 执行查询,结果保存在result中
    return result

@DFF.API('查询函数3', category='query')
def my_query_3(start_time):
    '''
    一个查询函数,只包含开始时间或结束时间
    '''
    # 执行查询,结果保存在result中
    return result

@DFF.API('查询函数4', category='query')
def my_query_4(end_time, start_time=None):
    '''
    一个查询函数,包含起止时间参数,且其中一项可选(也可以两者都可选)
    根据Python要求,可选参数必须在必选参数后面,所以此处start_time防止在end_time之后
    '''
    # 执行查询,结果保存在result中
    return result

@DFF.API('查询函数5', category='query')
def my_query_5(param1=None, param2=None):
    '''
    一个查询函数,没有起止时间,但附带自定义参数
    '''
    # 执行查询,结果保存在result中
    return result
 
@DFF.API('查询函数6', category='query')
def my_query_6(start_time, end_time, param1=None, param2=None):
    '''
    一个查询函数,包含起止时间,并附带自定义参数
    起止时间同样允许只存在其一,允许其中之一或两者可选
    '''
    # 执行查询,结果保存在result中
    return result

检测函数

检测函数主要用于DataFlux Studio中的高级检测功能。没有特定的参数和返回值要求,规范如下:

  • 分类category为check
  • 可以不附带或附带一个或多个可选参数
  • 不需要返回值

示例如下:

@DFF.API('检测函数1', category='check')
def my_check_1():
    '''
    一个检测函数,没有参数和返回值要求
    '''
    pass

@DFF.API('检测函数2', category='check')
def my_check_2(param1=None, param2=None):
    '''
    一个检测函数,没有参数和返回值要求
    '''
    pass

调用另一个脚本中的函数

Python代码可以根据功能、用途等不同需求划分到不同的脚本或脚本集中。位于不同脚本的代码是允许相互调用的。

需要调用另一个脚本中的函数时,只需要导入对应脚本即可。

导入另一个脚本时,必须按照固定写法:

# import <脚本集引用名>__<脚本引用名>
import demo__demo2

# 或 import <脚本集引用名>__<脚本引用名> as 别名
import demo__demo2 as demo2

也可以在脚本编辑器中,可以将鼠标指向脚本项右侧的问号,直接复制导入语句。

函数引用

脚本或脚本集并不是Python中的模块本身并不能通过import语句导入。但DataFlux.f(x)内部实现了动态加载机制,并允许使用import语句加载动态代码,因此,以下写法都是错误的。

# 错误写法1:将脚本集当作模块导入
import demo

# 错误写法2:将脚本当作模块导入
import demo.demo2
from demo import demo2

# 错误写法3:只导入某个函数
from demo.demo2 import some_func
from demo__demo2 import some_func

此外,在导入脚本时,应当注意不要产生循环引用,如:

# 脚本集 demo 下的脚本 demo1
import demo__demo2

# 脚本集 demo 下的脚本 demo2
import demo__demo3

# 脚本集 demo 下的脚本 demo3
import demo__demo1

注意,在同时编辑多个脚本时,如果当前导入了另一个脚本,那么被引用的脚本实际会以已发布的版本执行。任何时候都不会导入草稿!

更多详细内容请参考《DataFlux.f(x) 包学包会》