新增审批流插件并补齐申请侧权限

backend/app/task/actions.py

@@ -4,8 +4,8 @@
 from backend.common.socketio.server import sio
 
 
-@sio.event
+@sio.event(namespace='/ws')
 async def task_worker_status(sid, data) -> None:  # noqa: ANN001
     """任务 Worker 状态事件"""
     worker = await run_in_threadpool(celery_app.control.ping)
-    await sio.emit('task_worker_status', worker, sid)
+    await sio.emit('task_worker_status', worker, sid, namespace='/ws')

backend/common/socketio/actions.py

@@ -9,3 +9,8 @@ async def task_notification(msg: str) -> None:
     :return:
     """
     await sio.emit('task_notification', {'msg': msg})
+
+
+async def workflow_message_notification(user_id: int, data: dict) -> None:
+    """审批流消息通知"""
+    await sio.emit('workflow_message', data, room=f'user:{user_id}', namespace='/ws')

backend/common/socketio/server.py

@@ -18,11 +18,10 @@
     async_mode='asgi',
     cors_allowed_origins=settings.CORS_ALLOWED_ORIGINS,
     cors_credentials=True,
-    namespaces=['/ws'],
 )
 
 
-@sio.event
+@sio.event(namespace='/ws')
 async def connect(sid, environ, auth) -> bool:
     """Socket 连接事件"""
     if not auth:
@@ -42,16 +41,18 @@ async def connect(sid, environ, auth) -> bool:
 
     try:
         with request_cycle_context({settings.TRACE_ID_REQUEST_HEADER_KEY: uuid.uuid4().hex}):
-            await jwt_authentication(token)
+            user = await jwt_authentication(token)
     except Exception as e:
         log.info(f'WebSocket 连接失败：{e!s}')
         return False
 
+    await sio.save_session(sid, {'user_id': user.id, 'session_uuid': session_uuid}, namespace='/ws')
+    await sio.enter_room(sid, f'user:{user.id}', namespace='/ws')
     await redis_client.sadd(settings.TOKEN_ONLINE_REDIS_PREFIX, session_uuid)
     return True
 
 
-@sio.event
+@sio.event(namespace='/ws')
 async def disconnect(sid) -> None:
     """Socket 断开连接事件"""
     await redis_client.spop(settings.TOKEN_ONLINE_REDIS_PREFIX)

backend/plugin/workflow/README.md

@@ -0,0 +1,3 @@
+# Workflow Plugin
+
+Enterprise workflow and approval plugin for FBA.

backend/plugin/workflow/__init__.py

@@ -0,0 +1 @@
+"""Workflow plugin."""

backend/plugin/workflow/api/__init__.py

@@ -0,0 +1 @@
+"""Workflow api."""

backend/plugin/workflow/api/router.py

@@ -0,0 +1,16 @@
+from fastapi import APIRouter
+
+from backend.core.conf import settings
+from backend.plugin.workflow.api.v1.category import router as category_router
+from backend.plugin.workflow.api.v1.definition import router as definition_router
+from backend.plugin.workflow.api.v1.instance import router as instance_router
+from backend.plugin.workflow.api.v1.message import router as message_router
+from backend.plugin.workflow.api.v1.task import router as task_router
+
+v1 = APIRouter(prefix=f'{settings.FASTAPI_API_V1_PATH}/workflow')
+
+v1.include_router(category_router, prefix='/category', tags=['审批流分类'])
+v1.include_router(definition_router, prefix='/definition', tags=['审批流定义'])
+v1.include_router(instance_router, prefix='/instance', tags=['审批流实例'])
+v1.include_router(task_router, prefix='/task', tags=['审批流任务'])
+v1.include_router(message_router, prefix='/message', tags=['审批流消息'])

backend/plugin/workflow/api/v1/__init__.py

@@ -0,0 +1 @@
+"""Workflow v1 api."""

backend/plugin/workflow/api/v1/category.py

@@ -0,0 +1,37 @@
+from fastapi import APIRouter, Depends, Path
+
+from backend.common.pagination import DependsPagination, PageData
+from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, response_base
+from backend.common.security.jwt import DependsJwtAuth
+from backend.common.security.permission import RequestPermission
+from backend.common.security.rbac import DependsRBAC
+from backend.database.db import CurrentSession, CurrentSessionTransaction
+from backend.plugin.workflow.schema.category import (
+    CreateWorkflowCategoryParam,
+    GetWorkflowCategoryDetail,
+    UpdateWorkflowCategoryParam,
+)
+from backend.plugin.workflow.service.category_service import workflow_category_service
+
+router = APIRouter()
+
+
+@router.get('', dependencies=[DependsJwtAuth, DependsPagination])
+async def get_categories(db: CurrentSession) -> ResponseSchemaModel[PageData[GetWorkflowCategoryDetail]]:
+    return response_base.success(data=await workflow_category_service.get_list(db=db))
+
+
+@router.post('', dependencies=[Depends(RequestPermission('workflow:category:add')), DependsRBAC])
+async def create_category(db: CurrentSessionTransaction, obj: CreateWorkflowCategoryParam) -> ResponseModel:
+    await workflow_category_service.create(db=db, obj=obj)
+    return response_base.success()
+
+
+@router.put('/{pk}', dependencies=[Depends(RequestPermission('workflow:category:edit')), DependsRBAC])
+async def update_category(
+    db: CurrentSessionTransaction,
+    obj: UpdateWorkflowCategoryParam,
+    pk: int = Path(),
+) -> ResponseModel:
+    count = await workflow_category_service.update(db=db, pk=pk, obj=obj)
+    return response_base.success() if count > 0 else response_base.fail()

backend/plugin/workflow/api/v1/definition.py

@@ -0,0 +1,87 @@
+from fastapi import APIRouter, Depends, Path, Query, Request
+
+from backend.common.pagination import DependsPagination, PageData
+from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, response_base
+from backend.common.security.jwt import DependsJwtAuth
+from backend.common.security.permission import RequestPermission
+from backend.common.security.rbac import DependsRBAC
+from backend.database.db import CurrentSession, CurrentSessionTransaction
+from backend.plugin.workflow.schema.definition import (
+    CreateWorkflowDefinitionParam,
+    GetWorkflowDefinitionDetail,
+    PreviewWorkflowFlowResponse,
+    UpdateWorkflowDefinitionParam,
+)
+from backend.plugin.workflow.service.definition_service import workflow_definition_service
+
+router = APIRouter()
+
+
+@router.get('', dependencies=[Depends(RequestPermission('workflow:definition:list')), DependsRBAC, DependsPagination])
+async def get_definitions(db: CurrentSession) -> ResponseSchemaModel[PageData[GetWorkflowDefinitionDetail]]:
+    return response_base.success(data=await workflow_definition_service.get_list(db=db))
+
+
+@router.get('/available', dependencies=[DependsJwtAuth, DependsPagination])
+async def get_available_definitions(db: CurrentSession) -> ResponseSchemaModel[PageData[GetWorkflowDefinitionDetail]]:
+    return response_base.success(data=await workflow_definition_service.get_available_list(db=db))
+
+
+@router.get('/available/{pk}', dependencies=[DependsJwtAuth])
+async def get_available_definition(db: CurrentSession, pk: int = Path()) -> ResponseSchemaModel[GetWorkflowDefinitionDetail]:
+    return response_base.success(data=await workflow_definition_service.get_available(db=db, pk=pk))
+
+
+@router.get('/available/{pk}/preview-flow', dependencies=[DependsJwtAuth])
+async def preview_available_definition_flow(
+    db: CurrentSession,
+    request: Request,
+    pk: int = Path(),
+    form_data: str | None = Query(default=None),
+) -> ResponseSchemaModel[PreviewWorkflowFlowResponse]:
+    definition = await workflow_definition_service.get_available(db=db, pk=pk)
+    payload = await workflow_definition_service.preview_flow(
+        db=db,
+        definition=definition,
+        user_id=request.user.id,
+        form_data=workflow_definition_service._parse_config(form_data),
+    )
+    return response_base.success(data=payload)
+
+
+@router.get('/{pk}', dependencies=[Depends(RequestPermission('workflow:definition:list')), DependsRBAC])
+async def get_definition(db: CurrentSession, pk: int = Path()) -> ResponseSchemaModel[GetWorkflowDefinitionDetail]:
+    return response_base.success(data=await workflow_definition_service.get(db=db, pk=pk))
+
+
+@router.get('/{pk}/preview-flow', dependencies=[Depends(RequestPermission('workflow:definition:list')), DependsRBAC])
+async def preview_definition_flow(
+    db: CurrentSession,
+    request: Request,
+    pk: int = Path(),
+    form_data: str | None = Query(default=None),
+) -> ResponseSchemaModel[PreviewWorkflowFlowResponse]:
+    definition = await workflow_definition_service.get(db=db, pk=pk)
+    payload = await workflow_definition_service.preview_flow(
+        db=db,
+        definition=definition,
+        user_id=request.user.id,
+        form_data=workflow_definition_service._parse_config(form_data),
+    )
+    return response_base.success(data=payload)
+
+
+@router.post('', dependencies=[Depends(RequestPermission('workflow:definition:add')), DependsRBAC])
+async def create_definition(db: CurrentSessionTransaction, obj: CreateWorkflowDefinitionParam) -> ResponseModel:
+    await workflow_definition_service.create(db=db, obj=obj)
+    return response_base.success()
+
+
+@router.put('/{pk}', dependencies=[Depends(RequestPermission('workflow:definition:edit')), DependsRBAC])
+async def update_definition(
+    db: CurrentSessionTransaction,
+    obj: UpdateWorkflowDefinitionParam,
+    pk: int = Path(),
+) -> ResponseModel:
+    count = await workflow_definition_service.update(db=db, pk=pk, obj=obj)
+    return response_base.success() if count > 0 else response_base.fail()

backend/plugin/workflow/api/v1/instance.py

@@ -0,0 +1,59 @@
+from fastapi import APIRouter, Path, Request
+
+from backend.common.pagination import DependsPagination, PageData
+from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, response_base
+from backend.common.security.jwt import DependsJwtAuth
+from backend.database.db import CurrentSession, CurrentSessionTransaction
+from backend.plugin.workflow.schema.instance import GetWorkflowInstanceDetail, StartWorkflowInstanceParam
+from backend.plugin.workflow.service.instance_service import workflow_instance_service
+
+router = APIRouter()
+
+
+@router.post('', dependencies=[DependsJwtAuth])
+async def start_instance(
+    db: CurrentSessionTransaction,
+    request: Request,
+    obj: StartWorkflowInstanceParam,
+) -> ResponseSchemaModel[GetWorkflowInstanceDetail]:
+    instance = await workflow_instance_service.start(db=db, obj=obj, user_id=request.user.id)
+    return response_base.success(data=instance)
+
+
+@router.get('/my-apply', dependencies=[DependsJwtAuth, DependsPagination])
+async def get_my_apply(db: CurrentSession, request: Request) -> ResponseSchemaModel[PageData[GetWorkflowInstanceDetail]]:
+    return response_base.success(data=await workflow_instance_service.get_my_apply(db=db, user_id=request.user.id))
+
+
+@router.get('/my-todo', dependencies=[DependsJwtAuth, DependsPagination])
+async def get_my_todo(db: CurrentSession, request: Request) -> ResponseSchemaModel[PageData[GetWorkflowInstanceDetail]]:
+    return response_base.success(data=await workflow_instance_service.get_my_todo(db=db, user_id=request.user.id))
+
+
+@router.get('/todo-count', dependencies=[DependsJwtAuth])
+async def get_todo_count(db: CurrentSession, request: Request) -> ResponseSchemaModel[int]:
+    return response_base.success(data=await workflow_instance_service.get_todo_count(db=db, user_id=request.user.id))
+
+
+@router.get('/{pk}', dependencies=[DependsJwtAuth])
+async def get_instance(db: CurrentSession, request: Request, pk: int = Path()) -> ResponseSchemaModel[GetWorkflowInstanceDetail]:
+    return response_base.success(data=await workflow_instance_service.get_detail(db=db, pk=pk, user_id=request.user.id))
+
+
+@router.post('/{pk}/withdraw', dependencies=[DependsJwtAuth])
+async def withdraw_instance(
+    db: CurrentSessionTransaction,
+    request: Request,
+    pk: int = Path(),
+) -> ResponseSchemaModel[GetWorkflowInstanceDetail]:
+    return response_base.success(data=await workflow_instance_service.withdraw(db=db, pk=pk, user_id=request.user.id))
+
+
+@router.post('/{pk}/urge', dependencies=[DependsJwtAuth])
+async def urge_instance(
+    db: CurrentSessionTransaction,
+    request: Request,
+    pk: int = Path(),
+) -> ResponseModel:
+    await workflow_instance_service.urge(db=db, pk=pk, user_id=request.user.id)
+    return response_base.success()

backend/plugin/workflow/api/v1/message.py

@@ -0,0 +1,26 @@
+from fastapi import APIRouter, Path, Request
+
+from backend.common.pagination import DependsPagination, PageData
+from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, response_base
+from backend.common.security.jwt import DependsJwtAuth
+from backend.database.db import CurrentSession, CurrentSessionTransaction
+from backend.plugin.workflow.schema.message import GetWorkflowMessageDetail
+from backend.plugin.workflow.service.message_service import workflow_message_service
+
+router = APIRouter()
+
+
+@router.get('', dependencies=[DependsJwtAuth, DependsPagination])
+async def get_messages(db: CurrentSession, request: Request) -> ResponseSchemaModel[PageData[GetWorkflowMessageDetail]]:
+    return response_base.success(data=await workflow_message_service.get_list(db=db, user_id=request.user.id))
+
+
+@router.get('/unread-count', dependencies=[DependsJwtAuth])
+async def get_unread_count(db: CurrentSession, request: Request) -> ResponseSchemaModel[int]:
+    return response_base.success(data=await workflow_message_service.unread_count(db=db, user_id=request.user.id))
+
+
+@router.put('/{pk}/read', dependencies=[DependsJwtAuth])
+async def mark_read(db: CurrentSessionTransaction, request: Request, pk: int = Path()) -> ResponseModel:
+    count = await workflow_message_service.mark_read(db=db, pk=pk, user_id=request.user.id)
+    return response_base.success() if count > 0 else response_base.fail()

backend/plugin/workflow/api/v1/task.py

@@ -0,0 +1,38 @@
+from fastapi import APIRouter, Path, Request
+
+from backend.common.response.response_schema import ResponseSchemaModel, response_base
+from backend.common.security.jwt import DependsJwtAuth
+from backend.database.db import CurrentSession, CurrentSessionTransaction
+from backend.plugin.workflow.schema.task import (
+    ApproveWorkflowTaskParam,
+    GetWorkflowTaskDetail,
+    RejectWorkflowTaskParam,
+)
+from backend.plugin.workflow.service.task_service import workflow_task_service
+
+router = APIRouter()
+
+
+@router.get('/{pk}', dependencies=[DependsJwtAuth])
+async def get_task(db: CurrentSession, request: Request, pk: int = Path()) -> ResponseSchemaModel[GetWorkflowTaskDetail]:
+    return response_base.success(data=await workflow_task_service.get(db=db, pk=pk, user_id=request.user.id))
+
+
+@router.post('/{pk}/approve', dependencies=[DependsJwtAuth])
+async def approve_task(
+    db: CurrentSessionTransaction,
+    request: Request,
+    obj: ApproveWorkflowTaskParam,
+    pk: int = Path(),
+) -> ResponseSchemaModel[GetWorkflowTaskDetail]:
+    return response_base.success(data=await workflow_task_service.approve(db=db, pk=pk, user_id=request.user.id, obj=obj))
+
+
+@router.post('/{pk}/reject', dependencies=[DependsJwtAuth])
+async def reject_task(
+    db: CurrentSessionTransaction,
+    request: Request,
+    obj: RejectWorkflowTaskParam,
+    pk: int = Path(),
+) -> ResponseSchemaModel[GetWorkflowTaskDetail]:
+    return response_base.success(data=await workflow_task_service.reject(db=db, pk=pk, user_id=request.user.id, obj=obj))

backend/plugin/workflow/crud/__init__.py

@@ -0,0 +1 @@
+"""Workflow crud package."""

backend/plugin/workflow/crud/crud_category.py

@@ -0,0 +1,31 @@
+from collections.abc import Sequence
+
+from sqlalchemy import Select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy_crud_plus import CRUDPlus
+
+from backend.plugin.workflow.model import WorkflowCategory
+from backend.plugin.workflow.schema.category import CreateWorkflowCategoryParam, UpdateWorkflowCategoryParam
+
+
+class CRUDWorkflowCategory(CRUDPlus[WorkflowCategory]):
+    async def get(self, db: AsyncSession, pk: int) -> WorkflowCategory | None:
+        return await self.select_model(db, pk)
+
+    async def get_all(self, db: AsyncSession) -> Sequence[WorkflowCategory]:
+        return await self.select_models_order(db, 'sort', 'asc')
+
+    async def get_select(self) -> Select:
+        return await self.select_order('sort', 'asc')
+
+    async def get_by_code(self, db: AsyncSession, code: str) -> WorkflowCategory | None:
+        return await self.select_model_by_column(db, code=code)
+
+    async def create(self, db: AsyncSession, obj: CreateWorkflowCategoryParam) -> None:
+        await self.create_model(db, obj)
+
+    async def update(self, db: AsyncSession, pk: int, obj: UpdateWorkflowCategoryParam) -> int:
+        return await self.update_model(db, pk, obj)
+
+
+workflow_category_dao: CRUDWorkflowCategory = CRUDWorkflowCategory(WorkflowCategory)