Skip to content

审批系统

功能介绍#

审批系统主要用于处理审批请求，插件开发者可以在处理审批请求逻辑中，将审批请求发送到不同的第三方系统，第三方系统在处理完审批请求后，可以将处理结果发送回来

实现思路#

  • 首先创建审批动作，指定接口Path, method和负责处理的审批系统插件

  • 在中间件arkid.core.approve_request_middleware中根据扫描审批动作，拦截HTTP Request，

    1. 如果某个审批动作没有创建审批请求，则创建审批请求，分发CREATE_APPROVE_REQUEST事件，将HTTP Request存储在审批请求中, 中断HTTP Request
    2. 如果某个审批动作已经创建审批请求，判断该审批请求状态，如果状态为通过，继续执行HTTP Request，如果状态为拒绝，中断HTTP Request

  • 在审批系统插件中监听CREATE_APPROVE_REQUEST事件，通过create_approve_request将审批请求发送到其他第三方系统处理

  • 其他第三方审批系统处理完审批请求后，可以将审批结果通过接口返回

    • 同意审批请求接口

      • path：/approve_requests/{{request_id}}/pass/
      • method: PUT
      • 处理函数：pass_approve_request_handler
      • 需要实现抽象方法: pass_approve_request

    • 拒绝审批请求接口

      • path：/approve_requests/{{request_id}}/deny/'
      • method: PUT
      • 处理函数：deny_approve_request_handler
      • 需要实现抽象方法: deny_approve_request

抽象方法#

基类定义#

arkid.core.extension.approve_system.ApproveSystemExtension (Extension) #

Source code in arkid/core/extension/approve_system.py 
class ApproveSystemExtension(Extension):

    TYPE = "approve_system"

    composite_schema_map = {}
    created_composite_schema_list = []
    composite_key = 'type'
    composite_model = TenantExtensionConfig

    @property
    def type(self):
        return ApproveSystemExtension.TYPE

    def load(self):
        self.listen_event(
            core_event.CREATE_APPROVE_REQUEST, self.create_approve_request
        )
        self.pass_path = self.register_api(
            f'/approve_requests/{{request_id}}/pass/',
            'PUT',
            self.pass_approve_request_handler,
            response=ApproveRequestOut,
        )
        self.deny_path = self.register_api(
            f'/approve_requests/{{request_id}}/deny/',
            'PUT',
            self.deny_approve_request_handler,
            response=ApproveRequestOut,
        )
        super().load()

    @abstractmethod
    def create_approve_request(self, event, **kwargs):
        """
        抽象方法
        Args:
            event (arkid.core.event.Event): 创建审批请求事件
        """
        pass

    @operation(roles=[TENANT_ADMIN, PLATFORM_ADMIN])
    def pass_approve_request_handler(self, request, request_id):
        approve_request = ApproveRequest.valid_objects.get(id=request_id)
        self.pass_approve_request(request, approve_request)
        approve_request.status = 'pass'
        approve_request.save()
        return ErrorDict(ErrorCode.OK)

    @abstractmethod
    def pass_approve_request(self, request, approve_request):
        """
        抽象方法
        Args:
            request (django.http.HttpRequest): HTTP 请求
            approve_request (arkid.core.models.ApproveRequest): 需要同意的审批请求
        """
        pass

    @operation(roles=[TENANT_ADMIN, PLATFORM_ADMIN])
    def deny_approve_request_handler(self, request, request_id):
        approve_request = ApproveRequest.valid_objects.get(id=request_id)
        self.deny_approve_request(request, approve_request)
        approve_request.status = 'deny'
        approve_request.save()
        return ErrorDict(ErrorCode.OK)

    @abstractmethod
    def deny_approve_request(self, request, approve_request):
        """
        抽象方法
        Args:
            request (django.http.HttpRequest): HTTP 请求
            approve_request (arkid.core.models.ApproveRequest): 需要拒绝的审批请求
        """
        pass

    def create_tenant_config(self, tenant, config, name, type):
        tenant_config = super().create_tenant_config(tenant, config, name, type)
        tenant_config.config["pass_request_url"] = self.pass_path
        tenant_config.config["deny_request_url"] = self.deny_path
        tenant_config.save()
        return tenant_config

    def register_approve_system_schema(self, schema, system_type):
        self.register_config_schema(schema, self.package + '_' + system_type)
        self.register_composite_config_schema(
            schema, system_type, exclude=['extension']
        )

composite_model (BaseModel) django-model #

TenantExtensionConfig(id, is_del, is_active, updated, created, tenant, extension, config, name, type)

Source code in arkid/core/extension/approve_system.py 
class TenantExtensionConfig(BaseModel):

    class Meta(object):
        verbose_name = _("插件运行时配置")
        verbose_name_plural = _("插件运行时配置")

    tenant = models.ForeignKey('core.Tenant', blank=False, on_delete=models.PROTECT, verbose_name=_('租户'))
    extension = models.ForeignKey('Extension', blank=False, on_delete=models.PROTECT, verbose_name=_('插件'))
    config = models.JSONField(blank=True, default=dict, verbose_name=_('Runtime Config','运行时配置'))
    name = models.CharField(max_length=128, default='', verbose_name=_('名称'))
    type = models.CharField(max_length=128, default='', verbose_name=_('类型'))

config: JSONField blank django-field #

Runtime Config

created: DateTimeField blank django-field nullable #

创建时间

extension: ForeignKey django-field #

插件

id: UUIDField django-field #

ID

is_active: BooleanField django-field #

是否可用

is_del: BooleanField django-field #

是否删除

name: CharField django-field #

名称

tenant: ForeignKey django-field #

租户

type: CharField django-field #

类型

updated: DateTimeField blank django-field nullable #

更新时间

create_approve_request(self, event, **kwargs) #

抽象方法

Parameters:

Name Type Description Default
event arkid.core.event.Event

创建审批请求事件

 required
Source code in arkid/core/extension/approve_system.py 
@abstractmethod
def create_approve_request(self, event, **kwargs):
    """
    抽象方法
    Args:
        event (arkid.core.event.Event): 创建审批请求事件
    """
    pass

create_tenant_config(self, tenant, config, name, type) #

创建运行时配置

Parameters:

Name Type Description Default
tenant Tenant

租户

 required
config dict

config

 required
name str

运行时配置名字

 required
type str

配置类型

 required

Returns:

Type Description
TenantExtensionConfig

创建的对象
Source code in arkid/core/extension/approve_system.py 
def create_tenant_config(self, tenant, config, name, type):
    tenant_config = super().create_tenant_config(tenant, config, name, type)
    tenant_config.config["pass_request_url"] = self.pass_path
    tenant_config.config["deny_request_url"] = self.deny_path
    tenant_config.save()
    return tenant_config

deny_approve_request(self, request, approve_request) #

抽象方法

Parameters:

Name Type Description Default
request django.http.HttpRequest

HTTP 请求

 required
approve_request arkid.core.models.ApproveRequest

需要拒绝的审批请求

 required
Source code in arkid/core/extension/approve_system.py 
@abstractmethod
def deny_approve_request(self, request, approve_request):
    """
    抽象方法
    Args:
        request (django.http.HttpRequest): HTTP 请求
        approve_request (arkid.core.models.ApproveRequest): 需要拒绝的审批请求
    """
    pass

load(self) #

抽象方法，插件加载的入口方法

Source code in arkid/core/extension/approve_system.py 
def load(self):
    self.listen_event(
        core_event.CREATE_APPROVE_REQUEST, self.create_approve_request
    )
    self.pass_path = self.register_api(
        f'/approve_requests/{{request_id}}/pass/',
        'PUT',
        self.pass_approve_request_handler,
        response=ApproveRequestOut,
    )
    self.deny_path = self.register_api(
        f'/approve_requests/{{request_id}}/deny/',
        'PUT',
        self.deny_approve_request_handler,
        response=ApproveRequestOut,
    )
    super().load()

pass_approve_request(self, request, approve_request) #

抽象方法

Parameters:

Name Type Description Default
request django.http.HttpRequest

HTTP 请求

 required
approve_request arkid.core.models.ApproveRequest

需要同意的审批请求

 required
Source code in arkid/core/extension/approve_system.py 
@abstractmethod
def pass_approve_request(self, request, approve_request):
    """
    抽象方法
    Args:
        request (django.http.HttpRequest): HTTP 请求
        approve_request (arkid.core.models.ApproveRequest): 需要同意的审批请求
    """
    pass

示例#

extension_root.com_longgui_approve_system_arkid.ApproveSystemArkIDExtension (ApproveSystemExtension) #

Source code in extension_root/com_longgui_approve_system_arkid/__init__.py 
class ApproveSystemArkIDExtension(ApproveSystemExtension):
    def load(self):
        super().load()
        self.register_api(
            f'/approve_requests/',
            'GET',
            self.list_tenant_approve_requests,
            response=List[ApproveRequestListItemOut],
            tenant_path=True,
        )
        self.register_approve_system_schema(
            ApproveSystemArkIDConfigSchema, 'approve_system_arkid'
        )
        self.register_front_pages([router_page, waiting_page, approved_page])
        self.register_front_routers(router, approve_manage_router)

    @operation(List[ApproveRequestListItemOut], roles=[TENANT_ADMIN, PLATFORM_ADMIN])
    @paginate(CustomPagination)
    def list_tenant_approve_requests(
        self, request, tenant_id: str, is_approved: str = ''
    ):
        package = 'com.longgui.approve.system.arkid'
        requests = ApproveRequest.valid_objects.filter(
            tenant=request.tenant, action__extension__package=package
        )
        if is_approved == 'true':
            requests = requests.exclude(status="wait")
        elif is_approved == 'false':
            requests = requests.filter(status="wait")
        return requests

    def create_approve_request(self, event, **kwargs):
        pass

    def pass_approve_request(self, request, approve_request):
        response = restore_approve_request(approve_request)
        logger.info(f'Restore approve request with result: {response}')

    def deny_approve_request(self, request, approve_request):
        pass

create_approve_request(self, event, **kwargs) #

抽象方法

Parameters:

Name Type Description Default
event arkid.core.event.Event

创建审批请求事件

 required
Source code in extension_root/com_longgui_approve_system_arkid/__init__.py 
def create_approve_request(self, event, **kwargs):
    pass

deny_approve_request(self, request, approve_request) #

抽象方法

Parameters:

Name Type Description Default
request django.http.HttpRequest

HTTP 请求

 required
approve_request arkid.core.models.ApproveRequest

需要拒绝的审批请求

 required
Source code in extension_root/com_longgui_approve_system_arkid/__init__.py 
def deny_approve_request(self, request, approve_request):
    pass

load(self) #

抽象方法，插件加载的入口方法

Source code in extension_root/com_longgui_approve_system_arkid/__init__.py 
def load(self):
    super().load()
    self.register_api(
        f'/approve_requests/',
        'GET',
        self.list_tenant_approve_requests,
        response=List[ApproveRequestListItemOut],
        tenant_path=True,
    )
    self.register_approve_system_schema(
        ApproveSystemArkIDConfigSchema, 'approve_system_arkid'
    )
    self.register_front_pages([router_page, waiting_page, approved_page])
    self.register_front_routers(router, approve_manage_router)

pass_approve_request(self, request, approve_request) #

抽象方法

Parameters:

Name Type Description Default
request django.http.HttpRequest

HTTP 请求

 required
approve_request arkid.core.models.ApproveRequest

需要同意的审批请求

 required
Source code in extension_root/com_longgui_approve_system_arkid/__init__.py 
def pass_approve_request(self, request, approve_request):
    response = restore_approve_request(approve_request)
    logger.info(f'Restore approve request with result: {response}')

评论