开发手册¶

如何调用已封装的API¶

在使用元素类时，可能会遇到：需要调用的API接口类存在但未被元素类封装的情况，此时，推荐通过元素类的api成员来访问相应的API接口类。

例如变量组件VariableAPI的API接口类中有batch_variable_value方法，作用是：

根据多个变量元素的多个变量名称查询变量对应值

未被封装，则可以如下方式调用：

[1]:

from deepfos.api.models.variable import VariableValueBatchDTO
from deepfos.element.variable import Variable as SysVariable

v = SysVariable(element_name='test_val', path='/test')

batch_get = v.api.external.batch_variable_value

此时，调用batch_get等同于调用API接口类中的batch_variable_value，且经过了variable初始化，通过查看batch_variable_value方法的定义：

@post('get/variable-value/batch')
def batch_variable_value(self, variableValueBatchDTOList: List[VariableValueBatchDTO]) -> List[VariableBatchVO]:
    """
    根据多个变量元素的多个变量名称查询变量对应值

    """
    return {'body': variableValueBatchDTOList}

可知，需要的输入是VariableValueBatchDTO类的List，查看该模型的定义：

class VariableValueBatchDTO(BaseModel):
    """Variable Value Batch DTO

    .. admonition:: 引用接口

        - **POST** ``/variable-api/get/variable-value/batch``
    """
    #: 元素名称
    elementName: Optional[str]
    #: 文件夹Id
    folderId: Optional[str]
    #: 变量名称
    nameList: List[str]
    #: 元素路径
    path: Optional[str]

则定义batch_list做如下赋值：

[2]:

batch_list = [VariableValueBatchDTO(elementName='test_val',
                                    path='/test',
                                    nameList=['test_global_text','test_global_num']),
              VariableValueBatchDTO(elementName='tv', path='/', nameList=['ttt'])]

打印结果：

[3]:

batch_get(batch_list)

[3]:

[VariableBatchVO(elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path='\\test\\', variableMemberList=[VariableMemberVO(description={}, name='test_global_text', scope=1, value='abcd', valueType=1), VariableMemberVO(description={}, name='test_global_num', scope=1, value='1123.00', valueType=15)])]

可见获得了VariableBatchVO类的列表返回，查询结果符合预期，调用成功。

如何封装API¶

如遇到JAVA的Swagger接口文档中存在，但deepfos的api包里却没有封装接口的情况，可以根据Swagger里的描述自行封装API接口类。

例如服务配置接口，描述如下:

serverConfigParams

接口地址 /variable-server1-0/server-config-params

请求方式 GET

consumes ``

produces ["*/*"]

接口描述 ``

请求参数

参数名称

参数说明

请求类型

是否必须

数据类型

schema

app

应用id

header

true

string

space

空间id

header

true

string

user

用户id

header

true

integer

响应状态

状态码

说明

schema

200

OK

401

Unauthorized

403

Forbidden

404

Not Found

响应参数

暂无

响应示例
object

则可较简易地定义如下类用以封装：

[4]:

from deepfos.api.base import DynamicRootAPI, get

class MySimpleAPI(DynamicRootAPI):
    """变量组件接口"""
    module_type = 'VAR'

    @get('server-config-params')
    def service_config_params(self):
        return {}

此处 module_type的赋值 VAR ，是变量组件的module_type缩写。

[5]:

# 提供版本号调用
api = MySimpleAPI(version='1.0')
api.service_config_params()

[5]:

{'api.spaceServer.name': 'http://space-server',
 'eureka.client.service-url.defaultZone': 'http://admin123:12345678@eureka/eureka',
 'api.appServer.name': 'http://app-server',
 'eureka.instance.ip-address': '172.16.3.97',
 'api.gateway.url.appServer': 'http://server-gateway/app-server',
 'api.gateway.url.spaceServer': 'http://server-gateway/space-server',
 'api.systemServer.name': 'http://system-server',
 'api.url.gateway': 'http://server-gateway'}

说明

API接口类一般继承于DynamicRootAPI和ChildAPI，如果将API调用的接口路径看作一个类似文件系统的目录，则应将根目录定义为继承DynamicRootAPI的类，定义了module_type作为根目录的标识，实际对应了组件模块，分属的子目录为继承ChildAPI的类，定义了该组件下不同接口地址的get和post方法。

如上API封装方式可以实现对API的调用，然而对于较复杂的请求和响应的接口，这样定义既不便于提供请求参数，也不便于解析响应。

因此，可以进一步定义API请求和响应的模型类，定义时一般参考相应接口的Swagger页面里记录的接口描述。

例如如下接口描述：

查询变量元素成员接口

接口地址 /variable-server1-0/variable-element/variable-list

请求方式 POST

consumes ["application/json"]

produces ["*/*"]

接口描述 ``

请求参数

参数名称

参数说明

请求类型

是否必须

数据类型

schema

app

应用id

header

true

string

space

空间id

header

true

string

user

用户id

header

true

integer

variableApiDTOList

variableApiDTOList

body

true

array

VariableApiDTO

schema属性说明

VariableApiDTO

参数名称

参数说明

请求类型

是否必须

数据类型

schema

elementName

元素名称

body

true

string

folderId

文件夹Id

body

true

string

path

元素路径

body

false

string

scope

变量类型 (传空查询全部,传1查询全局变量,传2查询用户变量)

body

false

integer(int32)

响应状态

状态码	说明	schema
200	OK	VariableElementVO
201	Created
401	Unauthorized
403	Forbidden
404	Not Found

响应参数

参数名称	参数说明	类型	schema
global	全局变量参数	array	VariableAllParamVO
user	用户变量参数	array	VariableAllParamVO

schema属性说明

VariableAllParamVO

参数名称	参数说明	类型	schema
description	变量描述	object
elementName	元素名称	string
elementType	元素类型	string
folderId	文件夹Id	string
path	文件夹Id	string
variableParamVOList	变量参数列表	array	VariableParamVO

VariableParamVO

参数名称	参数说明	类型
description	变量描述	object
name	变量名称	string
scope	变量类型 1全局,2用户	integer(int32)
value	变量值	string
valueType	变量类型(字段类型)	integer(int32)

响应示例

{
  "global": [
      {
          "description": {},
          "elementName": "csc",
          "elementType": "VAR",
          "folderId": 0,
          "path": 0,
          "variableParamVOList": [
              {
                  "description": {},
                  "name": "变量1",
                  "scope": 1,
                  "value": "",
                  "valueType": 0
              }
          ]
      }
  ],
  "user": [
      {
          "description": {},
          "elementName": "csc",
          "elementType": "VAR",
          "folderId": 0,
          "path": 0,
          "variableParamVOList": [
              {
                  "description": {},
                  "name": "变量1",
                  "scope": 1,
                  "value": "",
                  "valueType": 0
              }
          ]
      }
  ]
}

可以在deepfos.api.models中定义如下模型类：

请求：

[6]:

from typing import *
from pydantic import Field

from deepfos.api.models.base import BaseModel

class VariableApiDTO(BaseModel):
    """Variable Api DTO

    .. admonition:: 引用接口

        - **POST** ``/variable-element/variable-list``
    """
    #: 元素名称
    elementName: str
    #: 文件夹Id
    folderId: str
    #: 元素路径
    path: Optional[str]
    #: 变量类型 (传空查询全部,传1查询全局变量,传2查询用户变量)
    scope: Optional[int]

响应：

[7]:

class Description(BaseModel):
    zh_cn: str = Field(None, alias='zh-cn')
    en: Optional[str]


class VariableParamVO(BaseModel):
    #: 变量描述
    description: Optional[Description]
    #: 变量名称
    name: str
    #: 变量类型 1全局,2用户
    scope: int
    #: 变量值
    value: Optional[str]
    #: 变量类型(字段类型)
    valueType: int


class VariableAllParamVO(BaseModel):
    #: 变量描述
    description: Optional[Description]
    #: 元素名称
    elementName: Optional[str]
    #: 元素类型
    elementType: Optional[str]
    #: 文件夹Id
    folderId: Optional[str]
    #: 文件夹Id
    path: Optional[str]
    #: 变量参数列表
    variableParamVOList: List[VariableParamVO]


class VariableElementVO(BaseModel):
    """Variable Element VO

    .. admonition:: 引用接口

        - **POST** ``/variable-element/variable-list`` (Response: 200)
    """
    #: 全局变量参数，global本身是一个关键字，因此定义为global_避免冲突，并用alias参数定义此处field名为global
    global_: List[VariableAllParamVO] = Field(..., alias='global')
    #: 用户变量参数
    user: List[VariableAllParamVO]

定义好模型类后，创建API类调用：

[8]:

from deepfos.api.base import post, ChildAPI
from deepfos.lib.decorator import cached_property

class ModelElement(ChildAPI):
    # 提供后接在组件地址的路径
    endpoint = '/variable-element'

    # 后接在组件地址和endpoint后的路径
    # post请求方式
    @post('variable-list')
    def list(self, detail_list):
        return {'body': detail_list}


class MyModelAPI(DynamicRootAPI):
    module_type = 'VAR'

    def variable(self):
        return ModelElement(self)

如上写法将变量组件内的API调用写入继承自ChildAPI的类中，作为变量组件类的方法回调，使用时可以通过 .variable() 的方式被调用，层次更为清晰。

[9]:

api = MyModelAPI(version='1.0')

detail = VariableApiDTO(elementName='test_val', path='/test', folderId='DIR27d7333f4da2')

api.variable().list(detail_list=[detail])

[9]:

{'global': [{'elementName': 'test_val',
   'elementType': 'VAR',
   'folderId': 'DIR3fd7682fb7f5',
   'path': '\\test\\',
   'variableParamVOList': [{'name': 'test_global_text',
     'scope': 1,
     'value': 'abcd',
     'valueType': 1,
     'description': {}},
    {'name': 'test_global_num',
     'scope': 1,
     'value': '1123.00',
     'valueType': 15,
     'description': {}}],
   'description': {}}],
 'user': [{'elementName': 'test_val',
   'elementType': 'VAR',
   'folderId': 'DIR3fd7682fb7f5',
   'path': '\\test\\',
   'variableParamVOList': [{'name': 'test_user_date',
     'scope': 2,
     'value': '2023-03-05 18:03:53',
     'valueType': 11,
     'description': {}},
    {'name': 'test_user_customlist',
     'scope': 2,
     'value': '1,3',
     'valueType': 13,
     'description': {}},
    {'name': 'test_user_text',
     'scope': 2,
     'value': None,
     'valueType': 1,
     'description': {'en': '', 'zh-cn': '测试中文描述'}},
    {'name': 'test_user_num',
     'scope': 2,
     'value': '999',
     'valueType': 15,
     'description': {}}],
   'description': {}}]}

可以看出，在定义了模型类后，依据VariableApiDTO类来赋值可以使请求的编辑更加准确，然而返回的仍然是字典，不是对象，此时在list方法上加上显式的参数和返回的类型声明后，可以使返回的字典被进一步处理为类对象：

[10]:

from typing import List

from deepfos.api.models.variable import VariableElementVO


class ModelElementPro(ChildAPI):
    endpoint = '/variable-element'

    # 显式声明，入参为VariableApiDTO类的列表，返回为VariableElementVO类的对象
    @post('variable-list')
    def list(self, detail_list: List[VariableApiDTO]) ->VariableElementVO:
        return {'body': detail_list}


class MyModelAPIPro(DynamicRootAPI):
    module_type = 'VAR'

    def variable(self):
        return ModelElementPro(self)

# 调用API类
api = MyModelAPIPro(version='1.0')

detail = VariableApiDTO(elementName='test_val',
                             path='/test',
                             folderId='DIR27d7333f4da2',
                             scope=1)

result = api.variable().list(detail_list=[detail])
type(result)

[10]:

deepfos.api.models.variable.VariableElementVO

为实现更好的复用性，可将MyModelAPIPro中的variable装饰为cached_property，则可在单个MyModelAPIPro中只在使用时初始化一次：

[11]:

from deepfos.lib.decorator import cached_property

class ModelElementPro(ChildAPI):
    endpoint = '/variable-element'

    # 显式声明，入参为VariableApiDTO类的列表，返回为VariableElementVO类的对象
    @post('variable-list')
    def list(self, detail_list: List[VariableApiDTO]) ->VariableElementVO:
        return {'body': detail_list}


class MyModelAPIPro(DynamicRootAPI):
    module_type = 'VAR'

    @cached_property
    def variable(self):
        return ModelElementPro(self)

如何扩展元素类¶

如果已有元素类内没有想要使用的逻辑，可以通过继承元素类的方式扩展逻辑。

例如在Variable类中增加查询变量元素成员接口的逻辑，将调用api的/variable-element/variable-list，即ElementAPI内的list方法，可通过如下继承实现：

[12]:

from deepfos.element.variable import Variable


class MyVariable(Variable):

    def list_element(self) -> VariableElementVO:
        # construct_from方法是deepfos的BaseModel类中提供的方法，
        # 可以传入model，从中提取变量与值的键值对为当前model赋值
        # 此处element_info是在ElementBase初始化时调用API赋值的，包含了元素基本信息
        DTO = VariableApiDTO.construct_from(self.element_info)
        return self.api.variable.list(variableApiDTOList=[DTO])

v = MyVariable(element_name='test_val', path='/test')

v.list_element()

[12]:

VariableElementVO(global_=[VariableAllParamVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path='\\test\\', variableParamVOList=[VariableParamVO(description={}, name='test_global_text', scope=1, value='abcd', valueType=1), VariableParamVO(description={}, name='test_global_num', scope=1, value='1123.00', valueType=15)])], user=[VariableAllParamVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path='\\test\\', variableParamVOList=[VariableParamVO(description={}, name='test_user_date', scope=2, value='2023-03-05 18:03:53', valueType=11), VariableParamVO(description={}, name='test_user_customlist', scope=2, value='1,3', valueType=13), VariableParamVO(description={'en': '', 'zh-cn': '测试中文描述'}, name='test_user_text', scope=2, value=None, valueType=1), VariableParamVO(description={}, name='test_user_num', scope=2, value='999', valueType=15)])])

补充¶

基于V1.1的新特性，针对同步元素的实现实际上是通过同步化异步元素来完成的，因此如上逻辑亦可通过继承实现AsyncVariable，实现其异步方法后，再同步化得到相应的同步元素类，这样做的好处，是在使用时，隐含了对并发执行特性的支持。

写法示例：

[13]:

from deepfos.element.variable import AsyncVariable
from deepfos.element.base import SyncMeta


class MyAsyncVariable(AsyncVariable):

    async def list_element(self) -> VariableElementVO:
        DTO = VariableApiDTO.construct_from(self.element_info)
        # 注意此处使用的是async_api,为异步实现的api实例
        return await self.async_api.variable.list(variableApiDTOList=[DTO])

class MyVariable(MyAsyncVariable, metaclass=SyncMeta):
    # 通过类成员synchronize来指定需同步化的异步方法
    # 此处需包括原来自AsyncVariable的save方法和新增的list_element方法
    synchronize = (
        'save',
        'list_element',
    )

v = MyVariable(element_name='test_val', path='/test')

v.list_element()

[13]:

VariableElementVO(global_=[VariableAllParamVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path='\\test\\', variableParamVOList=[VariableParamVO(description={}, name='test_global_text', scope=1, value='abcd', valueType=1), VariableParamVO(description={}, name='test_global_num', scope=1, value='1123.00', valueType=15)])], user=[VariableAllParamVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path='\\test\\', variableParamVOList=[VariableParamVO(description={}, name='test_user_date', scope=2, value='2023-03-05 18:03:53', valueType=11), VariableParamVO(description={}, name='test_user_customlist', scope=2, value='1,3', valueType=13), VariableParamVO(description={'en': '', 'zh-cn': '测试中文描述'}, name='test_user_text', scope=2, value=None, valueType=1), VariableParamVO(description={}, name='test_user_num', scope=2, value='999', valueType=15)])])

如何封装元素类¶

元素类内的方法和函数（即在deepfos.element包内类的函数），本质上都是对deepfos.api包内API接口方法的调用和包装，因此，如果遇到API类存在，但没有被封装成元素类时，可以调用deepfos.api内相应元素的API接口方法来实现对元素的操作，进而封装成元素类。

建议

为了将组织接口路径的过程统一化，在编写调用API的类时，建议继承基类ElementBase，并绑定api_class为当前所需操作的API类，以保证接口路径组织正确。

如下即为封装了查询变量名列表方法的元素类样例：

[14]:

from deepfos.api.variable import VariableAPI
from deepfos.api.models.variable import VariableElementVO,VariableNameListVO
from deepfos.element.base import ElementBase

class MyVariableElement(ElementBase):
    api_class = VariableAPI
    api:VariableAPI

    def list_variable(self,scope:int=None)->VariableNameListVO:
        # 注意此处使用的是api,为同步实现的api实例
        return self.api.external.list_variable(elementName=self.element_info.elementName,
                                               folderId=self.element_info.folderId,
                                               scope=scope)

v = MyVariableElement(element_name='test_val', path='/test')
v.list_variable()

[14]:

VariableNameListVO(global_=[VariableNameVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_global_text'), VariableNameVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_global_num')], user=[VariableNameVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_user_date'), VariableNameVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_user_customlist'), VariableNameVO(description={'en': '', 'zh-cn': '测试中文描述'}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_user_text'), VariableNameVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_user_num')])

补充¶

同样可通过先实现异步类及其异步方法的方式，来获得V1.1特性的支持。

写法示例：

[15]:

class MyAsyncVariableElement(ElementBase):
    api_class = VariableAPI

    async def list_variable(self,scope:int=None)->VariableNameListVO:
        # 注意此处使用的是async_api,为异步实现的api实例
        return await self.async_api.external.list_variable(elementName=self.element_info.elementName,
                                               folderId=self.element_info.folderId,
                                               scope=scope)

class MyVariableElement(MyAsyncVariableElement, metaclass=SyncMeta):
    synchronize = (
        'list_variable',
    )

v = MyVariableElement(element_name='test_val', path='/test')
v.list_variable()

[15]:

VariableNameListVO(global_=[VariableNameVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_global_text'), VariableNameVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_global_num')], user=[VariableNameVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_user_date'), VariableNameVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_user_customlist'), VariableNameVO(description={'en': '', 'zh-cn': '测试中文描述'}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_user_text'), VariableNameVO(description={}, elementName='test_val', elementType='VAR', folderId='DIR3fd7682fb7f5', path=None, status=True, variableName='test_user_num')])

参数名称	参数说明	请求类型	是否必须	数据类型
app	应用id	header	true	string
space	空间id	header	true	string
user	用户id	header	true	integer

参数名称	参数说明	请求类型	是否必须	数据类型
elementName	元素名称	body	true	string
folderId	文件夹Id	body	true	string
path	元素路径	body	false	string
scope	变量类型 (传空查询全部,传1查询全局变量,传2查询用户变量)	body	false	integer(int32)