目录 |
---|
类型 | 基类 | 简介 |
---|---|---|
过滤条件 | DynamicConditionBase | 用于动态过滤条件 |
校验器 | DynamicCheckBase | 用于数据新增或编辑时的校验 |
模型插件 | BaseModelPlugIn | 用于数据新增或编辑的前后处理插件 |
list_plugin | BaseListPlugIn | 用于对list接口返回的数据进行二次处理 |
动态Action | DynamicActionBase | 用于按钮中调用云函数 |
数据检查 | DynamicValidateCheckBase | 用于数据规范性的检查 |
报表公式 | BaseFormulaObject | 用于报表计算 |
薪酬公式 | BaseSalaryFormulaObject | 用于薪酬计算 |
社保公式 | BaseSocialFormulaObject | 用于社保计算 |
工作流云函数 | WorkFlowDynamicScriptBase | 用于工作流程 |
同步 | ThirdBaseUpdateObject | |
个性化短信服务 | MobileMessageServiceBase | |
休假标准公式 | LeaveStandardCalcDynamicBase | |
公共个性化公式 | BaseCustomerFormulaObject | |
其他 | object |
信息 |
---|
在云函数中调用接口使用self.call_open_api('hcm.model.get',{'id'_:1,'model':'Employee'}) 或者 self.call_open_api({name='hcm.model.get','id'_:1,params='model':'Employee'}) 下面的使用方式会报错: self.call_open_api('hcm.model.get',{api_name='id'_:1,api_params='model':'Employee'}) |
1.过滤条件
1.名称命名规则
公司ID_condition_云函数名称 例如 127_condition_test
2.案例
代码块 |
---|
class DynamicCondition(DynamicConditionBase):
"""
用于列表添加动态过滤条件
使用示例:"conditions": {"job_info.employee_category_id": "dynamic#employee_category_dynamic_condition"}
"""
def condition(self, query_, field_, condition_context):
result = self.call_open_api("hcm.model.list", param={"model": "emp_role_detail", "filter_dict": {
"employee_id": self.plugin_context.employee.id}})['list']
job_step_value = [0]
if len(result) > 0:
if result[0]['emp_type'] == 1: # 是什么类型人员
job_step_value = [546, 547, 548]
elif result[0]['emp_type'] == 2:
job_step_value = [549, 550, 551]
else:
job_step_value = [0]
query_ = query_.filter(field_.in_(job_step_value))
return query_ |
3.云函数测试参数
key | 含义 | 案例 | ||
---|---|---|---|---|
model | 模型 | Employee | ||
state | 场景 | all | ||
params | 参数 | 对应模型的获取list接口的参数
| ||
condition | 动态过滤 |
| ||
virtual_employee_id | 虚拟身份 | 可指定人员,若未填写则默认为当前用户 |
2.校验器
1.名称命名规则
DataChecker_xxx
2.案例
代码块 |
---|
class EndBeginTimeChecker(DynamicCheckBase):
"""
云函数校验器示例,在对象新增,修改时会校验
"""
def check(self, info):
result = {}
result['success'] = True
return result |
3.云函数测试参数
key | 含义 | 案例 |
---|---|---|
model | 模型 | Employee |
params | 参数 | 对应模型的增加或编辑info的参数 {"info":{"name":"1"}} |
virtual_employee_id | 虚拟身份 | 可指定人员,若未填写则默认为当前用户 |
3.模型插件
1.名称命名规则
无
2.案例
代码块 |
---|
class ModelPlugInXXX(BaseModelPlugIn):
"""
模型插件
应用范围:create_data,edit_data,remove_data,remove_datas
如果类型为pre,调用pre方法,类型为post,调用post方法
"""
def pre(self, company_id, **kwargs):
"""
Pre插件
kwargs:经过转换的原函数调用参数,经过干预后,返回kwargs继续调用原函数
"""
return kwargs
def post(self, company_id, result,**kwargs):
"""
Post插件
result:原函数调用返回的结果,经过干预后,返回
kwargs:原函数调用参数
"""
return result |
3.云函数测试参数
key | 含义 | 案例 | ||||
---|---|---|---|---|---|---|
model | 模型 | Employee | ||||
params | 参数 | 编辑:
新增:
| ||||
plugin | 插件 | 类型分为:pre_create_data,post_create_data,pre_edit_data,pre_edit_data
| ||||
virtual_employee_id | 虚拟身份 | 可指定人员,若未填写则默认为当前用户 |
4.list_plugin
1.命名规则
无
2.案例
代码块 |
---|
class ListPlugInXXX(BaseListPlugIn):
"""
模型data list插件
应用范围:对list接口返回的数据进行二次处理
"""
def post(self, context, _list, _count, plugin_config, **kwargs):
for _l in _list:
_id = _l.get('id')
_l['name']='aaaa'
logging.info(_id)
return _list, _count |
3.云函数测试参数
key | 含义 | 案例 | ||
---|---|---|---|---|
model | 模型 | Employee | ||
state | 场景 | all | ||
params | 参数 | 对应模型的获取list接口的参数
| ||
list插件 |
| |||
virtual_employee_id | 虚拟身份 | 可指定人员,若未填写则默认为当前用户 |
5.动态Action
1.命名规则
公司ID_模型名称_云函数名称 例如 127_testModel_testAction
2.案例
代码块 |
---|
class VirtualAction(DynamicActionBase):
"""
动态Action方法,装成模型上的方法
使用方法
"""
def action(self, method_context, params):
# raise NotImplementedError
logging.info(method_context)
logging.info("in method")
logging.info(params)
return params |
3.云函数测试参数
key | 含义 | 案例 | ||
---|---|---|---|---|
model | 模型 | Employee | ||
params | 参数 |
| ||
virtual_employee_id | 虚拟身份 | 可指定人员,若未填写则默认为当前用户 |
6.数据检查
1.命名规则
无
2.案例
代码块 |
---|
class XXXXChecher(DynamicValidateCheckBase):
"""
数据检查,主要是检查数据逻辑关系是否有问题,返回值是 ['xxxxx1数据有问题','xxxx2数据有问题']
"""
def check(self, **kwargs):
error_msg = []
model = self.get_model('EmployeeEducation')
rows = self.db.query(model).filter(model.company_id == self.context.company.id).all()
emp_group = {}
for row in rows:
if len(emp_educations) == 0:
error_msg.append('人员:{}的最高学历为空'.format(emp))
elif len(emp_educations) > 1:
error_msg.append('人员:{}的最高学历大于一个'.format(emp))
return error_msg |
3.云函数测试参数
key | 含义 | 案例 |
---|---|---|
model | 模型 | Employee |
params | 参数 | 云函数中接收的参数 |
virtual_employee_id | 虚拟身份 | 可指定人员,若未填写则默认为当前用户 |
7.报表公式
1.命名规则
FlexFormula_xxx
2.案例
代码块 |
---|
class DynamicFunctionReport(BaseFormulaObject):
"""
报表云函数
"""
def do(self,dept_id):
# 引入日志
company_id = self.context.get('COMPANY_ID')
return {} |
3.云函数测试参数
key | 含义 | 案例 |
---|---|---|
报表过滤参数 | {"dept_id":1234} | |
params | 参数 | 云函数中接收的参数 |
virtual_employee_id | 虚拟身份 | 可指定人员,若未填写则默认为当前用户 |
8.薪酬公式
1.命名规则
SalaryFormula_xxx
2.案例
代码块 |
---|
class BaseSalaryFormula(BaseSalaryFormulaObject):
"""
薪酬公式云函数示例
"""
def execute(self):
_cache_key = 'my_cache_key'
_cache = self.executor.context.get(_cache_key)
if _cache is None:
employee_ids = [_p.get('employee_id') for _p in self.context['profile_list']]
#你的取数逻辑
self.executor.context[_cache_key] = _cache
_result = _cache.get(self.context['profile']['employee_id'], 0)
return _result |
3.云函数测试参数
key | 含义 | 案例 |
---|---|---|
profile_id | 社保档案ID | |
month | 期间 | 对应的月份 |
params | 参数 | 云函数中接收的参数 |
virtual_employee_id | 虚拟身份 | 可指定人员,若未填写则默认为当前用户 |
9.社保公式
1.命名规则
SocialFormula__xxx
2.案例
代码块 |
---|
class BaseSocialFormula(BaseSocialFormulaObject):
"""
社保公式云函数示例
"""
def execute(self):
_cache_key = 'my_cache_key'
_cache = self.executor.context.get(_cache_key)
if _cache is None:
employee_ids = [_p.get('employee_id') for _p in self.context['profile_list']]
#你的取数逻辑
self.executor.context[_cache_key] = _cache
_result = _cache.get(self.context['profile']['employee_id'], 0)
return _result |
3.云函数测试参数
key | 含义 | 案例 |
---|---|---|
profile_id | 社保档案ID | 1234 |
month | 期间 | 对应的月份 |
params | 参数 | 云函数中接收的参数 |
virtual_employee_id | 虚拟身份 | 可指定人员,若未填写则默认为当前用户 |
10.工作流云函数
1.命名规则
无
2.案例
代码块 |
---|
class WorkFlowDynamicScriptHandle(WorkFlowDynamicScriptBase):
def execute(self, **kwargs):
"""
方案很合理可能多项目使用,可以联系下流程平台作为经典解决方案案例
:param kwargs:
:return:
"""
# 1. 原则上禁止重复接口多次调用,禁止使用 for 循环调用接口 . (不得不多次调用除外)
# 2. 尽量不使用 try,如果是校验可以 raise
# 3. 云函数标准耗时原则耗时不超 1秒
# 4. log 测试完成后需要移除
return |
3.云函数测试参数
key | 含义 | 案例 |
---|---|---|
params | 参数 | 云函数中接收的参数 |
virtual_employee_id | 虚拟身份 | 可指定人员,若未填写则默认为当前用户 |
11.同步
1.命名规则
third_data_sync_xxx(云函数里面的__third_type__)
12.个性化短信服务
待更新
13.公共个性化公式
命名规则:CommonFormula_xxx
案例:
xxx例如设置成GET_MES_D,则里面公式固定拼接成C_xxx, 其中:C_ 为平台公共个性化标识
云函数命名为: CommonFormula_GET_MES_D
公式命名为:C_GET_MES_D
详细示例注释:
'''
取MES模型某车间,某天,某班组,某岗位的某指标的数值 模型: U_MES_XJ_FFJ_01 --------公式编辑器的注释内容
C_GET_MES_D($('档案参数.车间(工段)编号'),$('档案参数.日期'),$('档案参数.班组'),$('档案参数.岗位'),'淀粉投入量')------公式以及公式传参定义内容
'''
公式会自动进行参数解析,然后可以进行公式个性化的逻辑处理
代码块 |
---|
class CommonFormulaGETMESD(BaseCustomerFormulaObject):
"""
公共个性化公式云函数示例
"""
def execute(self, u_department_code, u_date, u_team_name, u_post_name, u_assess_desc, **kwargs):
"""
取MES模型某车间,某天,某班组,某岗位的某指标的数值 模型: U_MES_XJ_FFJ_01
C_GET_MES_D($('档案参数.车间(工段)编号'),$('档案参数.日期'),$('档案参数.班组'),$('档案参数.岗位'),'淀粉投入量')
"""
param = {
'model': 'U_MES_XJ_FFJ_01',
'filter_dict': {'u_date': u_date,
'u_post_name': u_post_name,
'u_department_code': u_department_code,
'u_team_name': u_team_name,
'u_assess_desc': u_assess_desc},
'extra_property': {
'only_list': True,
'fields': [
{'field': ['u_quanity_num']}
]
}
}
_list = CustomerUtil.call_open_api("hcm.model.list", param)["list"]
if _list:
result = _list[0]["u_quanity_num"]
else:
result = None
return result
|
效果:
14.动态OpenAPI
1.新建私有动态API
如图所示
建议描述部分包含以下内容:
- 云函数实现的业务
- 云函数参数和返回结果的说明
- 测试参数
2.测试私有动态API
3.使用方式
- 内部使用和标准的OPEN API无异
- 外部使用需要结合访问令牌使用, 如下
- 外部使用方式有两种, 如下
15.云函数工具类
1. DataUtil
convert_num_to_id
作用:用于编码换取ID,解决云函数中写死固定编码而无法迁移问题(业务数据编码保证唯一)
可换取的模型类别:参数 含义 是否必需 model_name 模型名称 是 num
编码 是 default
默认值,当查询不到时如果存在默认值可以不抛出异常,返回设定的默认值 否 - 组织岗位类('OrgDepartment', 'BaseDepartment', 'Department', 'OrgUnit', 'Unit', 'OrgPosition','Position')
基于DepartmentHistory,返回origin_id,即组织ID,DepartmentHistory不支持获取本身原始ID行为。 - 流程(Business),返回流程ID
- 码表(common_basic_item_data.性别),返回码表ID
- 实体(EmployeeCateory, PositionStatus 等),返回实际数据ID
- 组织岗位类('OrgDepartment', 'BaseDepartment', 'Department', 'OrgUnit', 'Unit', 'OrgPosition','Position')
代码块 | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
class TestNumToId(BasePrivateApiService):
"""
动态私有OpenAPI
使用方法
"""
def execute(self, **kwargs):
"""
参数:number
default 设置为None时,如果编码换取不到ID时,程序不会抛出异常
"""
position_id = DataUtil.convert_num_to_id('OrgPosition', kwargs['number'], default=None) # 组织类,岗位ID
employee_category_id = DataUtil.convert_num_to_id('EmployeeCategory', kwargs['number'], default=None) # 实体类,用工类型ID
business_id = DataUtil.convert_num_to_id('Business', kwargs['number'], default=None) # 流程,流程ID
item_id = DataUtil.convert_num_to_id('common_basic_item_data.性别', kwargs['number'], default=None) # 码表,代码项ID;如果是树形代码项,需要保证number唯一
return |
2.CustomerUtil
call_open_api
调用系统API
3.ModelUtil
HTML |
---|
<table>
<thead>
<tr>
<th>方法</th>
<th>描述</th>
</tr>
</thead>
<tbody>
<tr>
<td>count</td>
<td>分组统计数值
:param model:
:param filter_dict:
:param state:
:param fields:
:param sorts:
:param extra_property:
:param query_str:
:param filter_str:
:param group_by:
:param count_by:
:param distinct_count_by:
:param sum_by:
:param avg_by:
:param min_by:
:param max_by:
:param page_index:
:param page_size:
:return:
</td>
</tr>
<tr>
<td>count_value</td>
<td>统计模型满足条件的个数,只有一个值
:param model:
:param filter_dict:
:param state:
:param fields:
:param sorts:
:param extra_property:
:param query_str:
:param filter_str:
:return:
</td>
</tr>
<tr>
<td>create</td>
<td>新增模型数据
:param model:
:param info:
:param role:
:return:
</td>
</tr>
<tr>
<td>create_batch</td>
<td>批量新增模型数据
:param model:
:param info_list:
:param role:
:return:
</td>
</tr>
<tr>
<td>edit</td>
<td>修改模型数据
:param model:
:param id_:
:param info:
:param role:
:return:
</td>
</tr>
<tr>
<td>edit_batch</td>
<td>批量修改数据
:param model:
:param edit_list:
:param role:
:return:
</td>
</tr>
<tr>
<td>get</td>
<td>获取模型数据
:param model_:
:param id_:
:param kwargs:
:return:
</td>
</tr>
<tr>
<td>list</td>
<td>获取列表满足条件的所有数据(数据量少的时候用)
:param model:
:param filter_dict:
:param state:
:param fields:
:param fields_key: 直接传字段列表,这边直接进行切割,
:param sorts:
:param extra_property:
:param query_str:
:param filter_str:
:param page_index:
:param page_size:
:return:
</td>
</tr>
<tr>
<td>list_ids</td>
<td>获取列表满足条件的所有数据id,直接返回id列表
:param model:
:param filter_dict:
:param state:
:param fields:
:param fields_key: 直接传字段列表,这边直接进行切割,
:param sorts:
:param extra_property:
:param query_str:
:param filter_str:
:param page_index:
:param page_size:
:return:
</td>
</tr>
<tr>
<td>record</td>
<td>日志记录
日志查看页面:/common_model_list?model=DynamicLogRecord
:param content: 内容
:param category
:param type_: 1正常 2 异常 3 成功
:param enabled: 是否执行
:return:
</td>
</tr>
<tr>
<td>remove</td>
<td>删除数据
:param model:
:param id_:
:param role:
:return:
</td>
</tr>
<tr>
<td>remove_batch</td>
<td>批量删除数据
:param model:
:param ids:
:param role:
:return:
</td>
</tr>
<tr>
<td>yield_count</td>
<td>aggr接口分批获取,
:param model:
:param filter_dict:
:param state:
:param fields:
:param sorts:
:param extra_property:
:param query_str:
:param filter_str:
:param group_by:
:param count_by:
:param distinct_count_by:
:param sum_by:
:param avg_by:
:param max_by:
:param min_by:
:param page_size:
:return:
</td>
</tr>
<tr>
<td>yield_list</td>
<td>循环按分页取列表,每次返回一个list 迭代
:param model:
:param filter_dict:
:param state:
:param fields:
:param fields_key: 接收fields key列表,['name','position.name']
:param sorts:
:param extra_property:
:param query_str:
:param filter_str:
:param page_size:
:return:
示例: for group in yield_list:
</td>
</tr>
</tbody>
</table>
<style type="text/css">
table,
td,
th {
border: 1px solid #000;
border-collapse: collapse;
}
table td {
white-space: break-spaces;
padding: 8px 16px;
}
thead >tr {
background: rgb(237 238 242);
height: 32px;
font-size: 18px;
}
tbody > tr:nth-of-type(even) {
background-color: rgb(237 238 242);
}
</style> |
董树森
...