应用sdk使用手册 - flagify-com/OctoMation GitHub Wiki
SDK在应用中提供对Octomation功能的简便调用,方便开发者操作Octomation的系统功能,实现更丰富的应用功能。
安装方法: pip install action-sdk-for-cache-mock
平台中已经内置了SDK,本地安装的action-sdk-for-cache-mock用于开发应用时使用,应用发布上传到系统时无需更改import的包
功能 | 描述 |
---|---|
上下文 | 获取执行的上下文信息(context_info),只能获取到当前执行事件内的信息。 |
日志 | 提供日志调试功能,可将日志上报到执行信息中 |
缓存键值对 | 提供事件维度的键值缓存 |
文件操作 | 提供文件保存/获取 |
若要使用context_info,则在方法参数上必须明确传入context_info,然后无需初始化直接使用即可。例如:
def function_name(params, assets,context_info)
功能 | 获取方法 | 数据类型 | 备注 |
---|---|---|---|
获取执行链 | context_info["activieId"] | string | 根据activieId获取执行信息 |
获取应用名称 | context_info["appName"] | string | 获取当前执行的应用名称 |
获取动作名称 | context_info["actionName"] | string | 获取当前执行的动作名称 |
获取事件ID | context_info["eventId"] | int | 获取当前事件ID |
判断日志模式 | context_info["logMode"] | bool | 判断当前是否是日志模式运行 |
获取执行人 | context_info["executor"] | string | 获取当前执行人 |
获取执行时间 | context_info["executeTime"] | int | 获取当前执行时间 |
在方法中使用SDK必须先要初始化,方法必须传入上下文信息参数context_info
。
from action_sdk_for_cache.action_cache_sdk import HoneyGuide
def function_name(params, assets,context_info):
hg_client = HoneyGuide(context_info=context_info)
功能 | 使用方法 | 备注 |
---|---|---|
打印日志 | hg_client.actionLog.info("测试") | 实例化HoneyGuide类时的参数是上游传下来的,所以实例化直接复制即可. |
缓存cache的作用域是事件,只有同一事件里执行的应用可以共享cache
功能 | 使用方法 | 数据类型 | 备注 |
---|---|---|---|
缓存键值对 | hg_client.keyCache.setString(key="lark_group", value="雾帜消防队") | str | key必须是字符串,不能为空,key的长度不能超过128,只能包含数字,字母,下划线并以字母开头; value如果不是字符串,转化成字符串后长度不能超过1024 |
获取key对应的键值 | hg_client.keyCache.getString(key="lark_group") | 如果key不存在则返回的是None |
功能 | 使用方法 | 数据类型 | 备注 |
---|---|---|---|
获取有权限的存储目录 | hg_client.fileCache.localFilePath() | 返回示例: /tmp/xxx | |
保存文件 | hg_client.fileCache.save("tmp/readme", keepDays=1) | save的入参是保存的文件,返回值为保存后文件的UUID,示例:5647a76e976e4c0ba52c9e0405197103。 keepDays参数,该参数非必传,不填默认永久保存 |
|
获取文件路径 | hg_client.fileCache.get("5647a76e976e4c0ba52c9e0405197103") | get的入参是文件的uuid,返回示例: /tmp/xxx | |
获取文件在前端的下载链接 | hg_client.fileCache.downloadUrl("5647a76e976e4c0ba52c9e0405197103") | downloadUrl的入参是要下载文件的uuid,返回示例: http://xxx/?/sss | |
判断文件是否存在 | result = hg_client.fileCache.isExist(fileUuid="5647a76e976e4c0ba52c9e0405197103") | 返回为boolean类型 |
功能 | 使用方法 | 数据类型 | 备注 |
---|---|---|---|
向作战室推送消息 | message = WarroomMessage(content="Hello", messageId="11111111", pushTimestamp=1593572373) hg_client.warroomMessageSender.send(message) |
messageId为字符串,字符串长度不能超过64,content为字符串,不能为空,pushTimestamp为int,其中messageId和pushTimestamp可以不传,content为必传参数,messageId可以重复,重复时仅展示pushTimestamp最大时的content |
功能 | 使用方法 | 值类型 | 是否抛异常 | 示例值 | 注意 |
---|---|---|---|---|---|
获取基础数据库密码 | ae_password = hg_client.dbConfig.aePassword() | string | 是 | 需要捕获ActionSdkException | |
获取应用库密码 | app_password = hg_client.dbConfig.appPassword() | string | 是 | ||
获取数据库的用户名 | username= hg_client.dbConfig.getUsername(domain="ae") | string | 否 | 目前数据库是只支持ae/app这两个库,domain参数不区分大小写 | |
获取数据库的host | host = hg_client.dbConfig.getHost(domain="ae") | string | 否 | 127.0.0.1 | 同上 |
获取数据库的端口 | port = hg_client.dbConfig.getPort(domain="ae") | string | 否 | "3306" | 同上 |
下面是对各个方法的使用示例,假设dev_action为开发的app中的一个action:
import os
import time
import shutil
import unittest
from unittest import TestCase
from action_sdk_for_cache.action_cache_sdk import HoneyGuide,WarroomMessage
def dev_action(params, assets, context_info):
"""
params/assets/context_info这三个参数都是上游传下来的,
不用自己手动传参
"""
hg_client = HoneyGuide(context_info=context_info)
# 上报日志, 这样在开启日志模式后就可以在事件处理处执行的动作中查看到日志信息
hg_client.actionLog.info(msg="测试")
# 缓存key, value.缓存某个变量的结果,以便后面的action可以获取到该结果
# (比如我多次执行某个动作,将多次执行的某个变量值做相关计算).缓存的数据必须可以json.dump
hg_client.keyCache.setString(key="cache_key", value="2")
# 获取刚才缓存key对应的value
hg_client.keyCache.getString(key="cache_key")
# 获取有权限的存储目录
validate_save_dir = hg_client.fileCache.localFilePath()
# 将文件保存至该目录下
filename = "readme.txt"
# 注意validate_save_dir不是一直不变的,所以不能通过该路径获取到上次
# 存储的文件进行操作,要想让前端下载文件,这里的存储操作是必须进行的
validate_save_path = os.path.join(validate_save_dir, filename)
with open(validate_save_path, "w", encoding="utf-8") as f:
f.write("Hey there!\n")
# 用sdk再保存一下文件,以便该文件能够在前端下载到.
file_uuid = hg_client.fileCache.save(filePath=validate_save_path)
# 获取sdk保存文件路径
file_path = hg_client.fileCache.get(fileUuid=file_uuid)
# 获取文件前端下载链接
download_url = hg_client.fileCache.downloadUrl(fileUuid=file_uuid)
print("first time download url: ", download_url)
##### 如果你想下次还用到这个文件,可以将刚才sdk保存文件的路径缓存起来
hg_client.keyCache.setString(key="last_save_file", value=file_path)
# 下次在该文件中新增内容
last_save_file_path = hg_client.keyCache.getString(key="last_save_file")
with open(last_save_file_path, "a+", encoding="utf-8") as f:
f.write("I'm new content!\n")
# 保存刚变更完的文件,用于前端下载
file_uuid = hg_client.fileCache.save(filePath=last_save_file_path)
download_url = hg_client.fileCache.downloadUrl(fileUuid=file_uuid)
print("second time download url: ", download_url)
# 往作战室推送消息
message = WarroomMessage(content="Hello", messageId="11111111", pushTimestamp=1593572373)
hg_client.warroomMessageSender.send(message)