iclientpy.server 源代码

from typing import List, Callable
from functools import partial
from iclientpy.codingui.distributedanalyst import get_datas_with_optionms
from .rest.apifactory import APIFactory
from .rest.api.servicespage import ServicesPage, ServiceComponentType
from .codingui.servicespage import get_services_by_component_type
from .rest.api.model import ServiceMetaInfo, UserEntity, RoleEntity, RolePermissions, UserInfo, AuthorizeSetting, \
    PostAuthorizeEntity, AuthorizeType, ServiceInstance
from iclientpy.codingui.comon import NamedObjects
from iclientpy.codingui.servicepublish import PrepareWorkspacePublish, PostWorkspaceExecutor
from iclientpy.codingui.servicespage import ui_class_register


[文档]class Server: _apifactory: APIFactory _services_page: ServicesPage _get_bigdatas: Callable def __init__(self, base_url: str, username: str = None, passwd: str = None, token: str = None, proxy: str = None, api_factory_clz=APIFactory): parms = (lambda ldict: {key: ldict[key] for key in ['base_url', 'username', 'passwd', 'token']})(locals()) parms['proxies'] = None if proxy is None else dict.fromkeys(['http', 'https'], proxy) self._apifactory = api_factory_clz(**parms) self._services_page = self._apifactory.servicespage() servicelist = self._services_page.list_services() # type: List[ServiceMetaInfo] bigdata_services = get_services_by_component_type(servicelist, ServiceComponentType.datacatalog, ServiceComponentType.distributedanalysis) if bigdata_services[ServiceComponentType.datacatalog] is not None and bigdata_services[ ServiceComponentType.distributedanalysis] is not None: datacatalog = self._apifactory.datacatalog_service(bigdata_services[ServiceComponentType.datacatalog].name) distributedanalysis = self._apifactory.distributedanalyst_service( bigdata_services[ServiceComponentType.distributedanalysis].name) self._get_bigdatas = partial(get_datas_with_optionms, distributedanalysis, datacatalog, self._apifactory.map_service) else: self._get_bigdatas = lambda: '服务不可用' @property def bigdatas(self): return self._get_bigdatas() @property def services(self): servicelist = self._services_page.list_services() # type: List[ServiceMetaInfo] result = NamedObjects() for meta_info in servicelist: ui = ui_class_register.new_service_ui_from_meta_info(meta_info, self._apifactory) if ui is not None: result[meta_info.name] = ui return result @property def service_names(self) -> List[str]: return [meta_info.name for meta_info in self._services_page.list_services()]
[文档] def prepare_workspace_for_publish(self): return PrepareWorkspacePublish(PostWorkspaceExecutor(self._apifactory), self._apifactory.management())
[文档] def get_users(self) -> List[List[str]]: """ 获取用户列表 Returns: 用户简略信息列表 """ mng = self._apifactory.security_management() return mng.get_users()
[文档] def get_user(self, name: str) -> UserInfo: """ 获取用户信息 Args: name: 用户名 Returns: 用户详细信息 """ mng = self._apifactory.security_management() return mng.get_user(name)
[文档] def create_user(self, name: str, password: str, roles: List[str] = None, description: str = None, user_groups: List[str] = None): """ 创建用户 Args: name: 用户名 password: 密码 roles: 角色 description: 描述信息 user_groups: 用户组 """ mng = self._apifactory.security_management() entity = UserEntity() entity.name = name entity.password = password entity.roles = roles entity.description = description entity.userGroups = user_groups result = mng.post_users(entity) if not result.succeed: raise Exception('创建用户失败')
[文档] def update_user(self, name: str, password: str = None, roles: List[str] = None, description: str = None, user_groups: List[str] = None): """ 更新用户信息 Args: name: 用户名 password: 密码 roles: 角色 description: 描述信息 user_groups: 用户组 """ mng = self._apifactory.security_management() entity = mng.get_user(name) entity.password = password if password is not None else entity.password entity.roles = roles if roles is not None else entity.roles entity.description = description if description is not None else entity.description entity.userGroups = user_groups if user_groups is not None else entity.userGroups result = mng.put_user(name, entity) if not result.succeed: raise Exception('更新用户失败')
[文档] def delete_users(self, names: List[str]): """ 批量删除用户 Args: names: 用户名列表 """ mng = self._apifactory.security_management() result = mng.put_users(names) if not result.succeed: raise Exception('删除用户失败')
[文档] def delete_user(self, name: str): """ 删除用户 Args: name: 用户名 """ mng = self._apifactory.security_management() result = mng.delete_user(name) if not result.succeed: raise Exception('删除用户失败')
[文档] def get_roles(self) -> List[RoleEntity]: """ 获取所有角色信息 Returns: 角色信息列表 """ mng = self._apifactory.security_management() return mng.get_roles()
[文档] def get_role(self, name: str) -> RoleEntity: """ 获取角色信息 Args: name: 角色名 Returns: 角色信息 """ mng = self._apifactory.security_management() return mng.get_role(name)
[文档] def create_role(self, name: str, users: List[str] = None, description: str = None, user_groups: List[str] = None, permissions: RolePermissions = None): """ 创建角色 Args: name: 角色名 users: 用户 description: 描述信息 user_groups: 用户组 permissions: 权限 """ mng = self._apifactory.security_management() entity = RoleEntity() entity.name = name entity.userGroups = user_groups entity.description = description entity.premissions = permissions entity.users = users result = mng.post_roles(entity) if not result.succeed: raise Exception('创建角色失败')
[文档] def update_role(self, name: str, users: List[str] = None, description: str = None, user_groups: List[str] = None, permissions: RolePermissions = None): """ 更新角色 Args: name: 角色名 users: 用户 description: 描述信息 user_groups: 用户组 permissions: 权限 """ mng = self._apifactory.security_management() entity = mng.get_role(name) entity.userGroups = user_groups if user_groups is not None else entity.userGroups entity.description = description if description is not None else entity.description entity.premissions = permissions if permissions is not None else entity.premissions entity.users = users if users is not None else entity.users result = mng.put_role(name, entity) if not result.succeed: raise Exception('更新角色失败')
[文档] def delete_role(self, name: str): """ 删除角色 Args: name: 角色名 """ mng = self._apifactory.security_management() result = mng.delete_role(name) if not result.succeed: raise Exception('删除角色失败')
[文档] def delete_roles(self, names: List[str]): """ 批量删除角色 Args: names: 角色名列表 """ mng = self._apifactory.security_management() result = mng.put_roles(names) if not result.succeed: raise Exception('删除角色失败')
[文档] def get_instances(self) -> List[ServiceInstance]: """ 获取所有服务实例 Returns: 服务实例信息列表 """ mng = self._apifactory.management() return mng.get_instances()
[文档] def get_instance(self, instance_name: str) -> ServiceInstance: """ 获取服务实例 Args: instance_name: 实例名称 Returns: 服务实例信息 """ mng = self._apifactory.management() return mng.get_instance(instance_name)
[文档] def grant_privileges_instances(self, instances_name: List[str], authorize_type: AuthorizeType, denied_roles: List[str] = None, permitted_roles: List[str] = None): """ 为服务示例授权 Args: instances_name: 服务实例名称列表 authorize_type: 验证类型 denied_roles: 禁止访问的角色列表 permitted_roles: 允许访问的角色列表 """ mng = self._apifactory.management() entity = PostAuthorizeEntity() entity.instances = instances_name entity.authorizeSetting = AuthorizeSetting() entity.authorizeSetting.type = authorize_type entity.authorizeSetting.deniedRoles = denied_roles entity.authorizeSetting.permittedRoles = permitted_roles result = mng.post_authorize(entity) if not result.succeed: raise Exception('授权失败')
[文档] def create_users_from_csv(self, path: str, name_key: str = 'name', password_key: str = 'password', roles_key: List[str] = 'roles', description_key: str = 'description', user_groups_key: List[str] = 'usergroups', sep=',', encoding: str = 'utf8'): """ 从csv文件里面读取用户信息,并在iServer上创建用户 Args: path: csv文件路径 name_key: 名称的列名 password_key: 密码的列名 roles_key: 角色的列名 description_key: 描述的列名 user_groups_key: 用户组的列名 sep: csv文件的分隔符 encoding: csv文件编码 """ with open(path, encoding=encoding) as csvfile: import csv reader = csv.DictReader(csvfile) for row in reader: name = row.get(name_key) password = row.get(password_key) roles = row.get(roles_key).split(sep) description = row.get(description_key) user_groups = row.get(user_groups_key).split(sep) self.create_user(name=name, password=password, roles=roles, description=description, user_groups=user_groups)