Feat: time period filter for workflow logs (#14271)

Signed-off-by: -LAN- <laipz8200@outlook.com>
Co-authored-by: -LAN- <laipz8200@outlook.com>

api/controllers/console/app/workflow_app_log.py

@@ -1,13 +1,18 @@
+from datetime import datetime
+
 from flask_restful import Resource, marshal_with, reqparse  # type: ignore
 from flask_restful.inputs import int_range  # type: ignore
+from sqlalchemy.orm import Session
 
 from controllers.console import api
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
+from extensions.ext_database import db
 from fields.workflow_app_log_fields import workflow_app_log_pagination_fields
 from libs.login import login_required
 from models import App
 from models.model import AppMode
+from models.workflow import WorkflowRunStatus
 from services.workflow_app_service import WorkflowAppService
 
 
@@ -24,14 +29,35 @@ class WorkflowAppLogApi(Resource):
         parser = reqparse.RequestParser()
         parser.add_argument("keyword", type=str, location="args")
         parser.add_argument("status", type=str, choices=["succeeded", "failed", "stopped"], location="args")
+        parser.add_argument(
+            "created_at__before", type=str, location="args", help="Filter logs created before this timestamp"
+        )
+        parser.add_argument(
+            "created_at__after", type=str, location="args", help="Filter logs created after this timestamp"
+        )
         parser.add_argument("page", type=int_range(1, 99999), default=1, location="args")
         parser.add_argument("limit", type=int_range(1, 100), default=20, location="args")
         args = parser.parse_args()
 
+        args.status = WorkflowRunStatus(args.status) if args.status else None
+        if args.created_at__before:
+            args.created_at__before = datetime.fromisoformat(args.created_at__before.replace("Z", "+00:00"))
+
+        if args.created_at__after:
+            args.created_at__after = datetime.fromisoformat(args.created_at__after.replace("Z", "+00:00"))
+
         # get paginate workflow app logs
         workflow_app_service = WorkflowAppService()
+        with Session(db.engine) as session:
             workflow_app_log_pagination = workflow_app_service.get_paginate_workflow_app_logs(
-            app_model=app_model, args=args
+                session=session,
+                app_model=app_model,
+                keyword=args.keyword,
+                status=args.status,
+                created_at_before=args.created_at__before,
+                created_at_after=args.created_at__after,
+                page=args.page,
+                limit=args.limit,
             )
 
             return workflow_app_log_pagination

api/controllers/service_api/app/workflow.py

@@ -1,7 +1,9 @@
 import logging
+from datetime import datetime
 
 from flask_restful import Resource, fields, marshal_with, reqparse  # type: ignore
 from flask_restful.inputs import int_range  # type: ignore
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import InternalServerError
 
 from controllers.service_api import api
@@ -25,7 +27,7 @@ from extensions.ext_database import db
 from fields.workflow_app_log_fields import workflow_app_log_pagination_fields
 from libs import helper
 from models.model import App, AppMode, EndUser
-from models.workflow import WorkflowRun
+from models.workflow import WorkflowRun, WorkflowRunStatus
 from services.app_generate_service import AppGenerateService
 from services.workflow_app_service import WorkflowAppService
 
@@ -125,14 +127,31 @@ class WorkflowAppLogApi(Resource):
         parser = reqparse.RequestParser()
         parser.add_argument("keyword", type=str, location="args")
         parser.add_argument("status", type=str, choices=["succeeded", "failed", "stopped"], location="args")
+        parser.add_argument("created_at__before", type=str, location="args")
+        parser.add_argument("created_at__after", type=str, location="args")
         parser.add_argument("page", type=int_range(1, 99999), default=1, location="args")
         parser.add_argument("limit", type=int_range(1, 100), default=20, location="args")
         args = parser.parse_args()
 
+        args.status = WorkflowRunStatus(args.status) if args.status else None
+        if args.created_at__before:
+            args.created_at__before = datetime.fromisoformat(args.created_at__before.replace("Z", "+00:00"))
+
+        if args.created_at__after:
+            args.created_at__after = datetime.fromisoformat(args.created_at__after.replace("Z", "+00:00"))
+
         # get paginate workflow app logs
         workflow_app_service = WorkflowAppService()
+        with Session(db.engine) as session:
             workflow_app_log_pagination = workflow_app_service.get_paginate_workflow_app_logs(
-            app_model=app_model, args=args
+                session=session,
+                app_model=app_model,
+                keyword=args.keyword,
+                status=args.status,
+                created_at_before=args.created_at__before,
+                created_at_after=args.created_at__after,
+                page=args.page,
+                limit=args.limit,
             )
 
             return workflow_app_log_pagination

api/fields/workflow_app_log_fields.py

@@ -17,8 +17,8 @@ workflow_app_log_partial_fields = {
 
 workflow_app_log_pagination_fields = {
     "page": fields.Integer,
-    "limit": fields.Integer(attribute="per_page"),
+    "limit": fields.Integer,
     "total": fields.Integer,
-    "has_more": fields.Boolean(attribute="has_next"),
-    "data": fields.List(fields.Nested(workflow_app_log_partial_fields), attribute="items"),
+    "has_more": fields.Boolean,
+    "data": fields.List(fields.Nested(workflow_app_log_partial_fields)),
 }

api/models/workflow.py

@@ -354,19 +354,6 @@ class WorkflowRunStatus(StrEnum):
     STOPPED = "stopped"
     PARTIAL_SUCCESSED = "partial-succeeded"
 
-    @classmethod
-    def value_of(cls, value: str) -> "WorkflowRunStatus":
-        """
-        Get value of given mode.
-
-        :param value: mode value
-        :return: mode
-        """
-        for mode in cls:
-            if mode.value == value:
-                return mode
-        raise ValueError(f"invalid workflow run status value {value}")
-
 
 class WorkflowRun(Base):
     """

api/services/workflow_app_service.py

@@ -1,30 +1,46 @@
 import uuid
+from datetime import datetime
 
-from flask_sqlalchemy.pagination import Pagination
-from sqlalchemy import and_, or_
+from sqlalchemy import and_, func, or_, select
+from sqlalchemy.orm import Session
 
-from extensions.ext_database import db
 from models import App, EndUser, WorkflowAppLog, WorkflowRun
 from models.enums import CreatedByRole
 from models.workflow import WorkflowRunStatus
 
 
 class WorkflowAppService:
-    def get_paginate_workflow_app_logs(self, app_model: App, args: dict) -> Pagination:
+    def get_paginate_workflow_app_logs(
+        self,
+        *,
+        session: Session,
+        app_model: App,
+        keyword: str | None = None,
+        status: WorkflowRunStatus | None = None,
+        created_at_before: datetime | None = None,
+        created_at_after: datetime | None = None,
+        page: int = 1,
+        limit: int = 20,
+    ) -> dict:
         """
-        Get paginate workflow app logs
-        :param app: app model
-        :param args: request args
-        :return:
+        Get paginate workflow app logs using SQLAlchemy 2.0 style
+        :param session: SQLAlchemy session
+        :param app_model: app model
+        :param keyword: search keyword
+        :param status: filter by status
+        :param created_at_before: filter logs created before this timestamp
+        :param created_at_after: filter logs created after this timestamp
+        :param page: page number
+        :param limit: items per page
+        :return: Pagination object
         """
-        query = db.select(WorkflowAppLog).where(
+        # Build base statement using SQLAlchemy 2.0 style
+        stmt = select(WorkflowAppLog).where(
             WorkflowAppLog.tenant_id == app_model.tenant_id, WorkflowAppLog.app_id == app_model.id
         )
 
-        status = WorkflowRunStatus.value_of(args.get("status", "")) if args.get("status") else None
-        keyword = args["keyword"]
         if keyword or status:
-            query = query.join(WorkflowRun, WorkflowRun.id == WorkflowAppLog.workflow_run_id)
+            stmt = stmt.join(WorkflowRun, WorkflowRun.id == WorkflowAppLog.workflow_run_id)
 
         if keyword:
             keyword_like_val = f"%{keyword[:30].encode('unicode_escape').decode('utf-8')}%".replace(r"\u", r"\\u")
@@ -40,20 +56,40 @@ class WorkflowAppService:
             if keyword_uuid:
                 keyword_conditions.append(WorkflowRun.id == keyword_uuid)
 
-            query = query.outerjoin(
+            stmt = stmt.outerjoin(
                 EndUser,
                 and_(WorkflowRun.created_by == EndUser.id, WorkflowRun.created_by_role == CreatedByRole.END_USER),
-            ).filter(or_(*keyword_conditions))
+            ).where(or_(*keyword_conditions))
 
         if status:
-            # join with workflow_run and filter by status
-            query = query.filter(WorkflowRun.status == status.value)
+            stmt = stmt.where(WorkflowRun.status == status)
 
-        query = query.order_by(WorkflowAppLog.created_at.desc())
+        # Add time-based filtering
+        if created_at_before:
+            stmt = stmt.where(WorkflowAppLog.created_at <= created_at_before)
 
-        pagination = db.paginate(query, page=args["page"], per_page=args["limit"], error_out=False)
+        if created_at_after:
+            stmt = stmt.where(WorkflowAppLog.created_at >= created_at_after)
 
-        return pagination
+        stmt = stmt.order_by(WorkflowAppLog.created_at.desc())
+
+        # Get total count using the same filters
+        count_stmt = select(func.count()).select_from(stmt.subquery())
+        total = session.scalar(count_stmt) or 0
+
+        # Apply pagination limits
+        offset_stmt = stmt.offset((page - 1) * limit).limit(limit)
+
+        # Execute query and get items
+        items = list(session.scalars(offset_stmt).all())
+
+        return {
+            "page": page,
+            "limit": limit,
+            "total": total,
+            "has_more": total > page * limit,
+            "data": items,
+        }
 
     @staticmethod
     def _safe_parse_uuid(value: str):

web/app/components/app/workflow-log/filter.tsx

@@ -2,11 +2,29 @@
 import type { FC } from 'react'
 import React from 'react'
 import { useTranslation } from 'react-i18next'
+import dayjs from 'dayjs'
+import { RiCalendarLine } from '@remixicon/react'
+import quarterOfYear from 'dayjs/plugin/quarterOfYear'
 import type { QueryParam } from './index'
 import Chip from '@/app/components/base/chip'
 import Input from '@/app/components/base/input'
+dayjs.extend(quarterOfYear)
 
-interface IFilterProps {
+const today = dayjs()
+
+export const TIME_PERIOD_MAPPING: { [key: string]: { value: number; name: string } } = {
+  1: { value: 0, name: 'today' },
+  2: { value: 7, name: 'last7days' },
+  3: { value: 28, name: 'last4weeks' },
+  4: { value: today.diff(today.subtract(3, 'month'), 'day'), name: 'last3months' },
+  5: { value: today.diff(today.subtract(12, 'month'), 'day'), name: 'last12months' },
+  6: { value: today.diff(today.startOf('month'), 'day'), name: 'monthToDate' },
+  7: { value: today.diff(today.startOf('quarter'), 'day'), name: 'quarterToDate' },
+  8: { value: today.diff(today.startOf('year'), 'day'), name: 'yearToDate' },
+  9: { value: -1, name: 'allTime' },
+}
+
+type IFilterProps = {
   queryParams: QueryParam
   setQueryParams: (v: QueryParam) => void
 }
@@ -27,6 +45,17 @@ const Filter: FC<IFilterProps> = ({ queryParams, setQueryParams }: IFilterProps)
           { value: 'stopped', name: 'Stop' },
         ]}
       />
+      <Chip
+        className='min-w-[150px]'
+        panelClassName='w-[270px]'
+        leftIcon={<RiCalendarLine className='h-4 w-4 text-text-secondary' />}
+        value={queryParams.period}
+        onSelect={(item) => {
+          setQueryParams({ ...queryParams, period: item.value })
+        }}
+        onClear={() => setQueryParams({ ...queryParams, period: '9' })}
+        items={Object.entries(TIME_PERIOD_MAPPING).map(([k, v]) => ({ value: k, name: t(`appLog.filter.period.${v.name}`) }))}
+      />
       <Input
         wrapperClassName='w-[200px]'
         showLeftIcon

web/app/components/app/workflow-log/index.tsx

@@ -4,21 +4,30 @@ import React, { useState } from 'react'
 import useSWR from 'swr'
 import { usePathname } from 'next/navigation'
 import { useDebounce } from 'ahooks'
+import { omit } from 'lodash-es'
+import dayjs from 'dayjs'
+import utc from 'dayjs/plugin/utc'
+import timezone from 'dayjs/plugin/timezone'
 import { Trans, useTranslation } from 'react-i18next'
 import Link from 'next/link'
 import List from './list'
-import Filter from './filter'
+import Filter, { TIME_PERIOD_MAPPING } from './filter'
 import Pagination from '@/app/components/base/pagination'
 import Loading from '@/app/components/base/loading'
 import { fetchWorkflowLogs } from '@/service/log'
 import { APP_PAGE_LIMIT } from '@/config'
 import type { App, AppMode } from '@/types/app'
+import { useAppContext } from '@/context/app-context'
+
+dayjs.extend(utc)
+dayjs.extend(timezone)
 
 export type ILogsProps = {
   appDetail: App
 }
 
 export type QueryParam = {
+  period: string
   status?: string
   keyword?: string
 }
@@ -48,7 +57,8 @@ const EmptyElement: FC<{ appUrl: string }> = ({ appUrl }) => {
 
 const Logs: FC<ILogsProps> = ({ appDetail }) => {
   const { t } = useTranslation()
-  const [queryParams, setQueryParams] = useState<QueryParam>({ status: 'all' })
+  const { userProfile: { timezone } } = useAppContext()
+  const [queryParams, setQueryParams] = useState<QueryParam>({ status: 'all', period: '2' })
   const [currPage, setCurrPage] = React.useState<number>(0)
   const debouncedQueryParams = useDebounce(queryParams, { wait: 500 })
   const [limit, setLimit] = React.useState<number>(APP_PAGE_LIMIT)
@@ -58,6 +68,13 @@ const Logs: FC<ILogsProps> = ({ appDetail }) => {
     limit,
     ...(debouncedQueryParams.status !== 'all' ? { status: debouncedQueryParams.status } : {}),
     ...(debouncedQueryParams.keyword ? { keyword: debouncedQueryParams.keyword } : {}),
+    ...((debouncedQueryParams.period !== '9')
+      ? {
+        created_at__after: dayjs().subtract(TIME_PERIOD_MAPPING[debouncedQueryParams.period].value, 'day').startOf('day').tz(timezone).format('YYYY-MM-DDTHH:mm:ssZ'),
+        created_at__before: dayjs().endOf('day').tz(timezone).format('YYYY-MM-DDTHH:mm:ssZ'),
+      }
+      : {}),
+    ...omit(debouncedQueryParams, ['period', 'status']),
   }
 
   const getWebAppType = (appType: AppMode) => {

web/app/components/workflow/nodes/_base/components/output-vars.tsx

@@ -42,7 +42,7 @@ export const VarItem: FC<VarItemProps> = ({
     <div className='py-1'>
       <div className='flex leading-[18px] items-center'>
         <div className='code-sm-semibold text-text-secondary'>{name}</div>
-        <div className='ml-2 system-xs-regular text-text-tertiary'>{type}</div>
+        <div className='ml-2 system-xs-regular text-text-tertiary capitalize'>{type}</div>
       </div>
       <div className='mt-0.5 system-xs-regular text-text-tertiary'>
         {description}

web/service/log.ts

@@ -17,7 +17,6 @@ import type {
   LogMessageAnnotationsResponse,
   LogMessageFeedbacksRequest,
   LogMessageFeedbacksResponse,
-  WorkflowLogsRequest,
   WorkflowLogsResponse,
   WorkflowRunDetailResponse,
 } from '@/models/log'
@@ -64,7 +63,7 @@ export const fetchAnnotationsCount: Fetcher<AnnotationsCountResponse, { url: str
   return get<AnnotationsCountResponse>(url)
 }
 
-export const fetchWorkflowLogs: Fetcher<WorkflowLogsResponse, { url: string; params?: WorkflowLogsRequest }> = ({ url, params }) => {
+export const fetchWorkflowLogs: Fetcher<WorkflowLogsResponse, { url: string; params: Record<string, any> }> = ({ url, params }) => {
   return get<WorkflowLogsResponse>(url, { params })
 }