Underground-Digital/Workflow-Engine

1import uuid2 3from flask_sqlalchemy.pagination import Pagination4from sqlalchemy import and_, or_5 6from extensions.ext_database import db7from models import App, EndUser, WorkflowAppLog, WorkflowRun8from models.enums import CreatedByRole9from models.workflow import WorkflowRunStatus10 11 12class WorkflowAppService:13    def get_paginate_workflow_app_logs(self, app_model: App, args: dict) -> Pagination:14        """15        Get paginate workflow app logs16        :param app: app model17        :param args: request args18        :return:19        """20        query = db.select(WorkflowAppLog).where(21            WorkflowAppLog.tenant_id == app_model.tenant_id, WorkflowAppLog.app_id == app_model.id22        )23 24        status = WorkflowRunStatus.value_of(args.get("status", "")) if args.get("status") else None25        keyword = args["keyword"]26        if keyword or status:27            query = query.join(WorkflowRun, WorkflowRun.id == WorkflowAppLog.workflow_run_id)28 29        if keyword:30            keyword_like_val = f"%{args['keyword'][:30]}%"31            keyword_conditions = [32                WorkflowRun.inputs.ilike(keyword_like_val),33                WorkflowRun.outputs.ilike(keyword_like_val),34                # filter keyword by end user session id if created by end user role35                and_(WorkflowRun.created_by_role == "end_user", EndUser.session_id.ilike(keyword_like_val)),36            ]37 38            # filter keyword by workflow run id39            keyword_uuid = self._safe_parse_uuid(keyword)40            if keyword_uuid:41                keyword_conditions.append(WorkflowRun.id == keyword_uuid)42 43            query = query.outerjoin(44                EndUser,45                and_(WorkflowRun.created_by == EndUser.id, WorkflowRun.created_by_role == CreatedByRole.END_USER),46            ).filter(or_(*keyword_conditions))47 48        if status:49            # join with workflow_run and filter by status50            query = query.filter(WorkflowRun.status == status.value)51 52        query = query.order_by(WorkflowAppLog.created_at.desc())53 54        pagination = db.paginate(query, page=args["page"], per_page=args["limit"], error_out=False)55 56        return pagination57 58    @staticmethod59    def _safe_parse_uuid(value: str):60        # fast check61        if len(value) < 32:62            return None63 64        try:65            return uuid.UUID(value)66        except ValueError:67            return None68