|
简道云API基本可以让简道云平台拥有无限的可能，也能充分实现开发者的思路和想法，让简道云平台更智能、更体贴、更个性。
V5版API是一个非常大的且很实用的更新，我这几天快速更新了V5版python语言的类，并且更新了原有的api代码，这真是一个大的工作量。
由于简道云用户多为简代码编程爱好者，时间、精力、能力都会受到限制，所以想着大家共享，共同完善。把我整理的类发出来，大家共同提建议，共同完善。
class APIUtils:
WEBSITE = "https://api.jiandaoyun.com"
RETRY_IF_LIMITED = True
# 构造函数
def __init__(self, api_key):
self.url_get_apps = APIUtils.WEBSITE + '/api/v5/app/list' # 用户应用查询接口
self.url_get_forms = APIUtils.WEBSITE + '/api/v5/app/entry/list' # 用户表单查询接口
self.url_get_widgets = APIUtils.WEBSITE + '/api/v5/app/entry/widget/list' # 表单字段查询接口
self.url_get_data = APIUtils.WEBSITE + '/api/v5/app/entry/data/list' # 查询多条数据接口
self.url_retrieve_data = APIUtils.WEBSITE + '/api/v5/app/entry/data/get' # 查询单条数据接口
self.url_update_data = APIUtils.WEBSITE + '/api/v5/app/entry/data/update' # 修改单条数据接口
self.url_data_batch_update = APIUtils.WEBSITE + '/api/v5/app/entry/data/batch_update' # 更新多条数据接口
self.url_create_data = APIUtils.WEBSITE + '/api/v5/app/entry/data/create' # 新建单条数据接口
self.url_batch_create_data = APIUtils.WEBSITE + '/api/v5/app/entry/data/batch_create' # 新建多条数据接口
self.url_delete_data = APIUtils.WEBSITE + '/api/v5/app/entry/data/delete' # 删除单条数据接口
self.url_batch_delete_data = APIUtils.WEBSITE + '/api/v5/app/entry/data/batch_delete' # 删除多条数据接口
self.url_get_upload_token = APIUtils.WEBSITE + '/api/v5/app/entry/file/get_upload_token' # 获取文件上传凭证和上传地址接口
self.url_workflow_approval_comments = APIUtils.WEBSITE + '/api/v1/app/{}/entry/{}/data/{}/approval_comments' # 该接口用来获取单条表单流程数据的审批意见
self.url_workflow_query = APIUtils.WEBSITE + '/api/v1/process/task/query' # 该接口用来获取用户当前待办信息
self.url_workflow_get = APIUtils.WEBSITE + '/api/v1/workflow/instance/get' # 该接口用来查询流程实例信息
self.url_workflow_close = APIUtils.WEBSITE + '/api/v1/workflow/instance/close' # 该接口用于管理员结束当前流程实例
self.url_workflow_list = APIUtils.WEBSITE + '/api/v1/workflow/task/list' # 该接口用来获取用户当前待办信息
self.url_workflow_approve = APIUtils.WEBSITE + '/api/v1/workflow/task/approve' # 该接口用来提交流程待办
self.url_workflow_rollback = APIUtils.WEBSITE + '/api/v1/workflow/task/rollback' # 该接口用来回退流程待办
self.url_workflow_transfer = APIUtils.WEBSITE + '/api/v1/workflow/task/transfer' # 该接口用来转交流程待办
self.url_get_department_data = APIUtils.WEBSITE + '/api/v5/corp/department/list' # （递归）获取部门列表
self.url_create_department = APIUtils.WEBSITE + '/api/v5/corp/department/create' # 创建部门
self.url_delete_department = APIUtils.WEBSITE + '/api/v5/corp/department/delete' # 删除部门
self.url_update_department = APIUtils.WEBSITE + '/api/v5/corp/department/update' # 修改部门
self.url_get_deptno_by_integrateid = APIUtils.WEBSITE + '/api/v5/corp/department/dept_no/get' # 根据集成模式通讯录的部门ID获取部门编号
self.url_import_department = APIUtils.WEBSITE + '/api/v5/corp/department/import' # 全量导入部门接口
self.url_get_user_retrieve = APIUtils.WEBSITE + '/api/v5/corp/user/get' # 获取成员的信息
self.url_create_user = APIUtils.WEBSITE + '/api/v5/corp/user/create' # 添加成员接口
self.url_update_user = APIUtils.WEBSITE + '/api/v5/corp/user/update' # 修改成员
self.url_get_member_list = APIUtils.WEBSITE + '/api/v5/corp/department/user/list' # （递归）获取部门成员
self.url_delete_user = APIUtils.WEBSITE + '/api/v5/corp/user/delete' # 删除成员
self.url_batch_delete_user = APIUtils.WEBSITE + '/api/v5/corp/user/batch_delete' # 批量删除成员
self.url_import_user = APIUtils.WEBSITE + '/api/v5/corp/user/import' # 增量导入成员接口
self.url_role_list = APIUtils.WEBSITE + '/api/v5/corp/role/list' # 列出角色接口
self.url_role_create = APIUtils.WEBSITE + '/api/v5/corp/role/create' # 创建一个自建角色
self.url_role_update = APIUtils.WEBSITE + '/api/v5/corp/role/update' # 更新一个自建角色
self.url_role_delete = APIUtils.WEBSITE + '/api/v5/corp/role/delete' # 删除一个自建角色
self.url_role_user_list = APIUtils.WEBSITE + '/api/v5/corp/role/user/list' # 列出角色下的成员
self.url_role_add_membersh = APIUtils.WEBSITE + '/api/v5/corp/role/add_membersh' # 为自建角色批量添加成员
self.url_role_remove_members = APIUtils.WEBSITE + '/api/v5/corp/role/remove_members' # 为自建角色批量移除成员
self.url_role_group_list = APIUtils.WEBSITE + '/api/v5/corp/role_group/list' # 列出自建角色组
self.url_role_group_create = APIUtils.WEBSITE + '/api/v5/corp/role_group/create' # 创建自建角色组
self.url_role_group_update = APIUtils.WEBSITE + '/api/v5/corp/role_group/update' # 更新自建角色组
self.url_role_group_delete = APIUtils.WEBSITE + '/api/v5/corp/role_group/delete' # 删除自建角色组
self.url_guest_department_list = APIUtils.WEBSITE + '/api/v5/corp/guest/department/list' # 列出我连接的企业
self.url_guest_member_list = APIUtils.WEBSITE + '/api/v5/corp/guest/member_list' # 列出我连接的企业的对接人
self.url_guest_user_get = APIUtils.WEBSITE + '/api/v5/corp/guest/user/get' # 获取我连接的企业对接人的详细信息
self.api_key = api_key
def get_req_header(self): # 带有认证信息的请求头
return {
'Authorization': 'Bearer ' + self.api_key,
'Content-Type': 'application/json;charset=utf-8'
}
def send_request(self, method, request_url, data): # 发送http请求
headers = self.get_req_header()
if method == 'GET':
res = requests.get(request_url, params=data, headers=headers, verify=False)
if method == 'POST':
res = requests.post(request_url, data=json.dumps(data), headers=headers, verify=False)
result = res.json()
# print("result:",result)
print("res:",res)
print("res.status_code:",res.status_code)
if res.status_code >= 400:
if result['code'] == 8303 and APIUtils.RETRY_IF_LIMITED:
# 5s后重试
print("频率过高")
time.sleep(8)
return self.send_request(method, request_url, data)
elif result['code'] == 4012 and APIUtils.RETRY_IF_LIMITED:
# 5s后重试
print('操作失败，当前表单正在执行其他批量编辑任务，请稍后重试')
time.sleep(8)
return self.send_request(method, request_url, data)
elif result['code'] == 1010:
print("不存在此用户")
pass
else:
raise Exception('请求错误！', result)
else:
return result
'''应用接口'''
def get_apps(self, limit, skip): # 获取应用
data = {
"limit": limit,
"skip": skip
}
result = self.send_request('POST', self.url_get_apps, data)
return result
'''表单和数据接口'''
def get_forms(self, appId): # 获取表单
data = {
"app_id": appId
}
result = self.send_request('POST', self.url_get_forms, data)
return result
def get_form_widgets(self, appId, entryId): # 获取表单字段
data = {
"app_id": appId,
"entry_id": entryId
}
result = self.send_request('POST', self.url_get_widgets, data)
return result['widgets']
def get_retrieve_data(self, appId, entryId, dataId): # 检索一条数据
data = {
"app_id": appId,
"entry_id": entryId,
'data_id': dataId
}
result = self.send_request('POST', self.url_retrieve_data, data)
return result['data']
def get_form_data(self, appId, entryId, dataId, limit, fields, data_filter): # 根据条件获取表单中的数据
result = self.send_request('POST', self.url_get_data, {
"app_id": appId,
"entry_id": entryId,
'data_id': dataId,
'limit': limit,
'fields': fields,
'filter': data_filter
})
# print("这是get_form_data根据查询参数查到的简道云表单的数据:",result)
return result['data']
def get_all_data(self, appId, entryId, fields, data_filter): # 获取表单中满足条件的所有数据
form_data = []
def get_next_page(dataId): # 递归取下一页数据
data = self.get_form_data(appId, entryId, dataId, 100, fields, data_filter)
if data:
for v in data:
form_data.append(v)
dataId = data[len(data) - 1]['_id']
get_next_page(dataId)
get_next_page('')
return form_data
def get_all_dataId(self, appId, entryId, data_filter): # 获取表单中满足条件的所有data_id，这个是我根据示例修改的，也应该是最稳定的
form_data = []
# 递归取下一页数据
def get_next_page(dataId):
data = self.get_form_data(appId, entryId, dataId, 100, None, data_filter)
if data:
for v in data:
form_data.append(v['_id'])
dataId = data[len(data) - 1]['_id']
get_next_page(dataId)
get_next_page('')
print("本次查询到的data_id条数为：", len(form_data))
return form_data
def get_dataId(self, appId, entryId, data_filter): # 根据字段的值查询dataid
form_dataId = []
result = self.get_form_data(appId, entryId, None, 100, None, data_filter)
if result:
for v in result:
temp_dataId = v['_id']
# print('v中的dataId:', temp_dataId)
form_dataId.append(temp_dataId)
# print('添加后的form_dataId:', form_dataId)
print("这是根据查询参数查到的简道云表单的dataID:", form_dataId)
return form_dataId
# 根据关键字段获取数据,这段代码是调试原来搜索错误的时候，想运用这一个进行纠错的，先放这儿，不删除
def get_dataId_from_dataid_and_filter(self, appId, entryId, data_id, data_filter): # 根据字段的值查询dataid
form_dataId = []
# print("这是get_dataId_from_dataid_and_filter根据查询参数查到的简道云表单的data_filter:", data_filter)
result = self.get_form_data(appId, entryId, data_id, 100, None, data_filter)
# print("这是get_dataId_from_dataid_and_filter根据查询参数查到的简道云表单的result:", result)
if result:
for v in result:
temp_dataId = v['_id']
# print('v中的dataId:', temp_dataId)
form_dataId.append(temp_dataId)
# print('添加后的form_dataId:', form_dataId)
print("这是get_dataId_from_dataid_and_filter根据查询参数查到的简道云表单的dataID:", len(form_dataId))
return form_dataId
def get_form_data_from_dataId(self, appId, entryId, dataId, limit):
data = {
"app_id": appId,
"entry_id": entryId,
'data_id': dataId,
'limit': limit
}
result = self.send_request('POST', self.url_get_data, data)
return result['data']
def create_data(self, appId, entryId, data): # 创建一条数据
transaction_id = self.get_transaction_id()
data = {
"app_id": appId,
"entry_id": entryId,
"transaction_id": transaction_id,
'data': data,
"is_start_trigger": True
}
result = self.send_request('POST', self.url_create_data, data)
return result['data']
def batch_create_data(self, appId, entryId, data_list): # 批量新增数据
transaction_id = self.get_transaction_id()
print("进入到了批量新增数据")
data = {
"app_id": appId,
"entry_id": entryId,
"transaction_id": transaction_id,
"data_list": data_list
}
result = self.send_request('POST', self.url_batch_create_data, data)
return result
def update_data(self, appId, entryId, dataId, data): # 更新数据
transaction_id = self.get_transaction_id()
data = {
"app_id": appId,
"entry_id": entryId,
"transaction_id": transaction_id,
'data_id': dataId,
'data': data,
'is_start_trigger': True
}
result = self.send_request('POST', self.url_update_data, data)
return result['data']
def batch_update_data(self, appId, entryId, data_ids, data): # 批量修改数据
transaction_id = self.get_transaction_id()
print("进入到了批量修改数据")
data = {
"app_id": appId,
"entry_id": entryId,
'transaction_id': transaction_id,
'data_ids': data_ids,
'data': data
}
result = self.send_request('POST', self.url_data_batch_update, data)
return result
def delete_data(self, appId, entryId, dataId): # 删除数据
data = {
"app_id": appId,
"entry_id": entryId,
'data_id': dataId,
"is_start_trigger": True
}
result = self.send_request('POST', self.url_delete_data, data)
return result
def batch_delete_data(self, appId, entryId, data_ids): # 删除数据
transaction_id = self.get_transaction_id()
data = {
"app_id": appId,
"entry_id": entryId,
'transaction_id': transaction_id,
'data_ids': data_ids
}
result = self.send_request('POST', self.url_batch_delete_data, data)
return result
'''条件修改，先根据filter的值进行查找，再一一修改'''
def update_from_filter(self, appId, entryId, data, data_filter):
# 判断查找的结果是否为空，如果不为空，则修改
form_dataId = self.get_all_dataId(appId, entryId, data_filter)
if form_dataId:
n = 100 # 将列表每100个组成一个小列表
for i in range(0, len(form_dataId), n):
dataIds = form_dataId[i:i + n] # 形成小列表，按100
self.batch_update_data(appId, entryId, dataIds, data)
# num_update = 0
# for dataId in form_dataId:
# print(dataId)
# self.update_data(appId, entryId, dataId, data)
# num_update += 1
# print("本次修改的数据条数：", num_update)
# return num_update
# 返回修改的条数
'''修改或新增，先根据data_filter的值进行查找，有则修改，无则新增'''
def update_or_create(self, appId, entryId, data, data_filter):
# 判断查找的结果是否为空，如果不为空，则修改
form_dataId = self.get_all_dataId(appId, entryId, data_filter)
if form_dataId:
num_update = 0
for dataId in form_dataId:
print(dataId)
self.update_data(appId, entryId, dataId, data)
num_update += 1
print("修改的数据数：", num_update)
return num_update
# 返回修改的条数
else:
# self.create_data(data)
result = self.create_data(appId, entryId, data)
print('新增数据：', result)
'''成员接口'''
def get_user_retrieve(self, username): # 获取成员的信息
data = {
"username": username
}
result = self.send_request('POST', self.url_get_user_retrieve, data)
return result["user"]
def get_user_department(self, username): # 获取成员的部门
data = {
"username": username
}
result = self.send_request('POST', self.url_get_user_retrieve, data)
return result["user"]['departments']
def create_user(self, username, name, departments): # 添加成员接口
data = {
"username": username,
"name": name,
"departments": departments
}
result = self.send_request('POST', self.url_create_user, data)
return result
def update_user_department_name(self, username, name, departments): # 更新成员部门和呢称
data = {
"username": username,
"name": name,
"departments": departments
}
result = self.send_request('POST', self.url_update_user, data)
print(result)
if result is None:
return
else:
return result["user"]
def delete_user(self, username): # 删除成员
data = {
"username": username
}
result = self.send_request('POST', self.url_delete_user, data)
return result
def batch_delete_user(self, usernames): # 批量删除成员
data = {
"usernames": usernames
}
result = self.send_request('POST', self.url_batch_delete_user, data)
return result
def import_user(self, users): # 批量导入成员
data = {
"users": users
}
result = self.send_request('POST', self.url_import_user, data)
return result
def get_member_list(self, department_id): # （递归）获取部门成员
data = {
"dept_no": department_id,
"has_child": True
}
result = self.send_request('POST', self.url_get_member_list, data)
return result["users"]
'''部门接口'''
def get_department_list(self, department_id): # （递归）获取部门列表
data = {
"dept_no": department_id,
"has_child": True
}
result = self.send_request('POST', self.url_get_department_data, data)
return result["departments"]
def create_department(self, data): # 创建部门
# data = {
# "name": "研发部门",
# "parent_no": 1,
# "dept_no": 2
# }
result = self.send_request('POST', self.url_create_department, data)
return result["department"]
def update_department(self, data): # 修改部门
# data = {
# "name": "研发部门",
# "parent_no": 1
# }
result = self.send_request('POST', self.url_update_department, data)
return result["department"]
def delete_department(self, dept_no): # 删除部门
data = {
"dept_no": dept_no
}
result = self.send_request('POST', self.url_delete_department, data)
return result["status"]
def get_deptno_by_integrateid(self, integrate_id): # 获取集成模式部门编号
data = {
"integrate_id": integrate_id
}
result = self.send_request('POST', self.url_get_deptno_by_integrateid, data)
return result["department"]
def import_department(self, departments): # 全量导入部门
data = {
"departments": departments
}
result = self.send_request('POST', self.url_import_department, data)
return result["status"]
'''角色接口'''
def list_role(self, skip, limit): # 列出角色
data = {
"skip": skip,
"limit": limit,
"has_internal": "Ture",
"has_sync": "Ture"
}
result = self.send_request('POST', self.url_role_list, data)
return result["roles"]
def create_role(self, group_no, name): # 只能创建内部自建角色。支持公共模式和集成模式
data = {
"group_no": group_no,
"name": name
}
result = self.send_request('POST', self.url_role_create, data)
return result["role"]
def update_role(self, data): # 对自建角色进行更新。支持公共模式和集成模式
# data = {
# "role_no":2558,
# "group_no": 2547,
# "name": "研发部门"
# }
result = self.send_request('POST', self.url_role_update, data)
return result["role"]
def delete_role(self, role_no): # 删除一个自建角色
data = {
"role_no": role_no
}
result = self.send_request('POST', self.url_role_delete, data)
return result["status"]
def list_user_role(self, skip, limit, role_no): # 删除一个自建角色
data = {
"skip": skip,
"limit": limit,
"role_no": role_no
}
result = self.send_request('POST', self.url_role_user_list, data)
return result["users"]
def add_membersh_role(self, role_no, usernames): # 删除一个自建角色
data = {
"role_no": role_no,
"usernames": usernames
}
result = self.send_request('POST', self.url_role_add_membersh, data)
return result["status"]
def remove_members_role(self, role_no, usernames): # 将成员从角色中批量移除。支持公共模式和集成模式
data = {
"role_no": role_no,
"usernames": usernames
}
result = self.send_request('POST', self.url_role_remove_members, data)
return result["status"]
'''角色组接口'''
def role_group_list(self, skip, limit): # 将角色组全部拉取出来。支持公共模式和集成模式。
data = {
"skip": skip,
"limit": limit,
"has_internal": "Ture",
"has_sync": "Ture"
}
result = self.send_request('POST', self.url_role_group_list, data)
return result["role_groups"]
def role_group_create(self, name): # 通过接口新建角色组。支持公共模式和集成模式。
data = {
"name": name
}
result = self.send_request('POST', self.url_role_group_create, data)
return result["role_group"]
def role_group_update(self, role_group_no, name): # 更新创建好的角色组信息。支持公共模式和集成模式。
data = {
"role_group_no": role_group_no,
"name": name
}
result = self.send_request('POST', self.url_role_group_update, data)
return result["role_group"]
def role_group_delete(self, role_group_no): # 更新创建好的角色组信息。支持公共模式和集成模式。
data = {
"role_group_no": role_group_no
}
result = self.send_request('POST', self.url_role_group_delete, data)
return result["status"]
'''企业互联接口'''
def guest_department_list(self, dept_no): # 能够获取所有外部部门的列表,dept_on空值，返回所有外部部门
data = {
"dept_no": dept_no
}
result = self.send_request('POST', self.url_guest_department_list, data)
return result["dept_list"]
def guest_member_list(self, dept_no): # 能够获取所有外部部门的列表,dept_on空值，返回全部
data = {
"dept_no": dept_no
}
result = self.send_request('POST', self.url_guest_member_list, data)
return result["member_list"]
def guest_user_get(self, username): # 获取某个 对接人 的信息详细信息。
data = {
"username": username
}
result = self.send_request('POST', self.url_guest_user_get, data)
return result["member"]
def get_transaction_id(self):
import uuid
uid = str(uuid.uuid1())
suid = ''.join(uid.split('-'))
return suid
|
