目录 |
---|
类型 | 基类 | 简介 |
---|---|---|
过滤条件 | 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'}) |
...
代码块 |
---|
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
...
代码块 |
---|
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
...
代码块 |
---|
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
...
代码块 |
---|
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
...
代码块 |
---|
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
...
代码块 |
---|
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
案例:
...
代码块 |
---|
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
新建私有动态API
如图所示
建议描述部分包含以下内容:
- 云函数实现的业务
- 云函数参数和返回结果的说明
- 测试参数
测试私有动态API
使用方式
- 内部使用和标准的OPEN API无异
- 外部使用需要结合访问令牌使用, 如下
- 外部使用方式有两种, 如下