diff --git a/api/controllers/console/app/workflow_statistic.py b/api/controllers/console/app/workflow_statistic.py index e72cd057a65..48661472584 100644 --- a/api/controllers/console/app/workflow_statistic.py +++ b/api/controllers/console/app/workflow_statistic.py @@ -1,28 +1,25 @@ -from flask import abort, jsonify +from datetime import datetime + from flask_restx import Resource from pydantic import BaseModel, Field, field_validator -from sqlalchemy.orm import sessionmaker +from werkzeug.exceptions import BadRequest from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models from controllers.console import console_ns from controllers.console.app.wraps import get_app_model +from controllers.console.flask_admission import console_account_admission from controllers.console.wraps import ( RBACPermission, RBACResourceScope, - account_initialization_required, model_validate, - rbac_permission_required, - setup_required, - with_current_user, ) -from extensions.ext_database import db +from extensions.ext_application_services import application_services from fields.base import ResponseModel from libs.datetime_utils import parse_time_range -from libs.login import login_required -from models.account import Account -from models.enums import WorkflowRunTriggeredFrom +from libs.helper import dump_response +from libs.login import current_account_with_tenant +from machinery.context import RequestContext from models.model import App, AppMode -from repositories.factory import DifyAPIRepositoryFactory class WorkflowStatisticQuery(BaseModel): @@ -83,13 +80,22 @@ register_response_schema_models( ) +def _resolve_statistic_time_range( + req_data: WorkflowStatisticQuery, +) -> tuple[datetime | None, datetime | None, str]: + timezone = current_account_with_tenant().account.timezone + assert timezone is not None + + try: + start_date, end_date = parse_time_range(req_data.start, req_data.end, timezone) + except ValueError as error: + raise BadRequest(str(error)) from error + + return start_date, end_date, timezone + + @console_ns.route("/apps//workflow/statistics/daily-conversations") class WorkflowDailyRunsStatistic(Resource): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - session_maker = sessionmaker(bind=db.engine, expire_on_commit=False) - self._workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker) - @console_ns.doc("get_workflow_daily_runs_statistic") @console_ns.doc(description="Get workflow daily runs statistics") @console_ns.doc(params={"app_id": "Application ID"}) @@ -99,41 +105,27 @@ class WorkflowDailyRunsStatistic(Resource): "Daily runs statistics retrieved successfully", console_ns.models[WorkflowDailyRunsStatisticResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @with_current_user - @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_MONITOR) + @console_account_admission( + rbac_resource_scope=RBACResourceScope.APP, + rbac_permission=RBACPermission.APP_MONITOR, + ) @get_app_model @model_validate(WorkflowStatisticQuery) - def get(self, req_data: WorkflowStatisticQuery, account: Account, app_model: App): - - assert account.timezone is not None - - try: - start_date, end_date = parse_time_range(req_data.start, req_data.end, account.timezone) - except ValueError as e: - abort(400, description=str(e)) - - response_data = self._workflow_run_repo.get_daily_runs_statistics( - tenant_id=app_model.tenant_id, + def get(self, req_data: WorkflowStatisticQuery, request_context: RequestContext, app_model: App): + start_date, end_date, timezone = _resolve_statistic_time_range(req_data) + response_data = application_services().workflow_statistics.get_daily_runs( + request_context, app_id=app_model.id, - triggered_from=WorkflowRunTriggeredFrom.APP_RUN, start_date=start_date, end_date=end_date, - timezone=account.timezone, + timezone=timezone, ) - return jsonify({"data": response_data}) + return dump_response(WorkflowDailyRunsStatisticResponse, {"data": response_data}) @console_ns.route("/apps//workflow/statistics/daily-terminals") class WorkflowDailyTerminalsStatistic(Resource): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - session_maker = sessionmaker(bind=db.engine, expire_on_commit=False) - self._workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker) - @console_ns.doc("get_workflow_daily_terminals_statistic") @console_ns.doc(description="Get workflow daily terminals statistics") @console_ns.doc(params={"app_id": "Application ID"}) @@ -143,41 +135,27 @@ class WorkflowDailyTerminalsStatistic(Resource): "Daily terminals statistics retrieved successfully", console_ns.models[WorkflowDailyTerminalsStatisticResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @with_current_user - @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_MONITOR) + @console_account_admission( + rbac_resource_scope=RBACResourceScope.APP, + rbac_permission=RBACPermission.APP_MONITOR, + ) @get_app_model @model_validate(WorkflowStatisticQuery) - def get(self, req_data: WorkflowStatisticQuery, account: Account, app_model: App): - - assert account.timezone is not None - - try: - start_date, end_date = parse_time_range(req_data.start, req_data.end, account.timezone) - except ValueError as e: - abort(400, description=str(e)) - - response_data = self._workflow_run_repo.get_daily_terminals_statistics( - tenant_id=app_model.tenant_id, + def get(self, req_data: WorkflowStatisticQuery, request_context: RequestContext, app_model: App): + start_date, end_date, timezone = _resolve_statistic_time_range(req_data) + response_data = application_services().workflow_statistics.get_daily_terminals( + request_context, app_id=app_model.id, - triggered_from=WorkflowRunTriggeredFrom.APP_RUN, start_date=start_date, end_date=end_date, - timezone=account.timezone, + timezone=timezone, ) - return jsonify({"data": response_data}) + return dump_response(WorkflowDailyTerminalsStatisticResponse, {"data": response_data}) @console_ns.route("/apps//workflow/statistics/token-costs") class WorkflowDailyTokenCostStatistic(Resource): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - session_maker = sessionmaker(bind=db.engine, expire_on_commit=False) - self._workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker) - @console_ns.doc("get_workflow_daily_token_cost_statistic") @console_ns.doc(description="Get workflow daily token cost statistics") @console_ns.doc(params={"app_id": "Application ID"}) @@ -187,41 +165,27 @@ class WorkflowDailyTokenCostStatistic(Resource): "Daily token cost statistics retrieved successfully", console_ns.models[WorkflowDailyTokenCostStatisticResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @with_current_user - @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_MONITOR) + @console_account_admission( + rbac_resource_scope=RBACResourceScope.APP, + rbac_permission=RBACPermission.APP_MONITOR, + ) @get_app_model @model_validate(WorkflowStatisticQuery) - def get(self, req_data: WorkflowStatisticQuery, account: Account, app_model: App): - - assert account.timezone is not None - - try: - start_date, end_date = parse_time_range(req_data.start, req_data.end, account.timezone) - except ValueError as e: - abort(400, description=str(e)) - - response_data = self._workflow_run_repo.get_daily_token_cost_statistics( - tenant_id=app_model.tenant_id, + def get(self, req_data: WorkflowStatisticQuery, request_context: RequestContext, app_model: App): + start_date, end_date, timezone = _resolve_statistic_time_range(req_data) + response_data = application_services().workflow_statistics.get_daily_token_costs( + request_context, app_id=app_model.id, - triggered_from=WorkflowRunTriggeredFrom.APP_RUN, start_date=start_date, end_date=end_date, - timezone=account.timezone, + timezone=timezone, ) - return jsonify({"data": response_data}) + return dump_response(WorkflowDailyTokenCostStatisticResponse, {"data": response_data}) @console_ns.route("/apps//workflow/statistics/average-app-interactions") class WorkflowAverageAppInteractionStatistic(Resource): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - session_maker = sessionmaker(bind=db.engine, expire_on_commit=False) - self._workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker) - @console_ns.doc("get_workflow_average_app_interaction_statistic") @console_ns.doc(description="Get workflow average app interaction statistics") @console_ns.doc(params={"app_id": "Application ID"}) @@ -231,29 +195,20 @@ class WorkflowAverageAppInteractionStatistic(Resource): "Average app interaction statistics retrieved successfully", console_ns.models[WorkflowAverageAppInteractionStatisticResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @with_current_user - @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_MONITOR) + @console_account_admission( + rbac_resource_scope=RBACResourceScope.APP, + rbac_permission=RBACPermission.APP_MONITOR, + ) @get_app_model(mode=[AppMode.WORKFLOW]) @model_validate(WorkflowStatisticQuery) - def get(self, req_data: WorkflowStatisticQuery, account: Account, app_model: App): - - assert account.timezone is not None - - try: - start_date, end_date = parse_time_range(req_data.start, req_data.end, account.timezone) - except ValueError as e: - abort(400, description=str(e)) - - response_data = self._workflow_run_repo.get_average_app_interaction_statistics( - tenant_id=app_model.tenant_id, + def get(self, req_data: WorkflowStatisticQuery, request_context: RequestContext, app_model: App): + start_date, end_date, timezone = _resolve_statistic_time_range(req_data) + response_data = application_services().workflow_statistics.get_average_app_interactions( + request_context, app_id=app_model.id, - triggered_from=WorkflowRunTriggeredFrom.APP_RUN, start_date=start_date, end_date=end_date, - timezone=account.timezone, + timezone=timezone, ) - return jsonify({"data": response_data}) + return dump_response(WorkflowAverageAppInteractionStatisticResponse, {"data": response_data}) diff --git a/api/extensions/ext_application_services.py b/api/extensions/ext_application_services.py index 6b10c8bcf54..d015bd651bb 100644 --- a/api/extensions/ext_application_services.py +++ b/api/extensions/ext_application_services.py @@ -27,6 +27,7 @@ from repositories.app_site_command_repository import AppSiteCommandRepository from repositories.data_source_api_key_auth_repository import SQLAlchemyDataSourceApiKeyAuthBindingRepository from repositories.data_source_oauth_binding_repository import SQLAlchemyDataSourceOAuthBindingRepository from repositories.explore_banner_query_repository import ExploreBannerQueryRepository +from repositories.factory import DifyAPIRepositoryFactory from repositories.installation_state_repository import InstallationStateRepository from repositories.oauth_server_repository import RedisOAuthServerTokenRepository, SQLAlchemyOAuthServerRepository from repositories.recommended_app_catalog_repository import DatabaseRecommendedAppCatalogRepository @@ -110,6 +111,7 @@ from services.webapp_access_query_service import ( WebAppAccessQueryService, WebAppAccessUnavailableError, ) +from services.workflow_statistic_query_service import WorkflowStatisticQueryService from services.workspace_member_query_service import WorkspaceMemberQueryService from services.workspace_member_role_resolver import DeploymentWorkspaceMemberRoleResolver from services.workspace_plan_gateway import DeploymentWorkspacePlanGateway @@ -173,6 +175,7 @@ class ApplicationServices: workspace_queries: WorkspaceQueryService workspace_member_queries: WorkspaceMemberQueryService tags: TagApplicationService + workflow_statistics: WorkflowStatisticQueryService def resolve_data_source_oauth(self, provider: str) -> DataSourceOAuthService: service = self.data_source_oauth.get(provider) @@ -407,6 +410,11 @@ def build_application_services( tags=TagApplicationService( tags=TagRepository(session_factory=database_client), ), + workflow_statistics=WorkflowStatisticQueryService( + workflow_runs=DifyAPIRepositoryFactory.create_api_workflow_run_repository( + session_maker=database_client, + ), + ), ) diff --git a/api/services/workflow_statistic_query_service.py b/api/services/workflow_statistic_query_service.py new file mode 100644 index 00000000000..4649fbe3aa8 --- /dev/null +++ b/api/services/workflow_statistic_query_service.py @@ -0,0 +1,97 @@ +"""Application service for workflow statistic queries.""" + +from datetime import datetime + +from machinery.context import RequestContext +from models.enums import WorkflowRunTriggeredFrom +from repositories.api_workflow_run_repository import APIWorkflowRunRepository +from repositories.types import ( + AverageInteractionStats, + DailyRunsStats, + DailyTerminalsStats, + DailyTokenCostStats, +) + + +class WorkflowStatisticQueryService: + def __init__(self, *, workflow_runs: APIWorkflowRunRepository) -> None: + self._workflow_runs = workflow_runs + + def get_daily_runs( + self, + context: RequestContext, + *, + app_id: str, + start_date: datetime | None, + end_date: datetime | None, + timezone: str, + ) -> list[DailyRunsStats]: + return self._workflow_runs.get_daily_runs_statistics( + tenant_id=self._workspace_id(context), + app_id=app_id, + triggered_from=WorkflowRunTriggeredFrom.APP_RUN, + start_date=start_date, + end_date=end_date, + timezone=timezone, + ) + + def get_daily_terminals( + self, + context: RequestContext, + *, + app_id: str, + start_date: datetime | None, + end_date: datetime | None, + timezone: str, + ) -> list[DailyTerminalsStats]: + return self._workflow_runs.get_daily_terminals_statistics( + tenant_id=self._workspace_id(context), + app_id=app_id, + triggered_from=WorkflowRunTriggeredFrom.APP_RUN, + start_date=start_date, + end_date=end_date, + timezone=timezone, + ) + + def get_daily_token_costs( + self, + context: RequestContext, + *, + app_id: str, + start_date: datetime | None, + end_date: datetime | None, + timezone: str, + ) -> list[DailyTokenCostStats]: + return self._workflow_runs.get_daily_token_cost_statistics( + tenant_id=self._workspace_id(context), + app_id=app_id, + triggered_from=WorkflowRunTriggeredFrom.APP_RUN, + start_date=start_date, + end_date=end_date, + timezone=timezone, + ) + + def get_average_app_interactions( + self, + context: RequestContext, + *, + app_id: str, + start_date: datetime | None, + end_date: datetime | None, + timezone: str, + ) -> list[AverageInteractionStats]: + return self._workflow_runs.get_average_app_interaction_statistics( + tenant_id=self._workspace_id(context), + app_id=app_id, + triggered_from=WorkflowRunTriggeredFrom.APP_RUN, + start_date=start_date, + end_date=end_date, + timezone=timezone, + ) + + @staticmethod + def _workspace_id(context: RequestContext) -> str: + workspace_id = context.active_workspace_id + if workspace_id is None: + raise RuntimeError("Console account admission did not resolve an active workspace") + return workspace_id diff --git a/api/tests/unit_tests/controllers/console/app/test_app_apis.py b/api/tests/unit_tests/controllers/console/app/test_app_apis.py index 42c5bd81f0b..c7f8286fac5 100644 --- a/api/tests/unit_tests/controllers/console/app/test_app_apis.py +++ b/api/tests/unit_tests/controllers/console/app/test_app_apis.py @@ -5,13 +5,14 @@ from __future__ import annotations import uuid from collections.abc import Iterator from inspect import unwrap +from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest from flask import Flask from pydantic import ValidationError from sqlalchemy import Engine -from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.orm import Session from werkzeug.exceptions import BadRequest, NotFound from controllers.console import console_ns @@ -70,7 +71,6 @@ from models import App, Site from models.account import Account, AccountStatus from models.engine import db from models.trigger import WorkflowWebhookTrigger -from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository from services.app_site_service import ( AppSiteAppNotFoundError, AppSiteChanges, @@ -106,6 +106,15 @@ def _make_app( return app +def _make_request_context() -> RequestContext: + return RequestContext( + request_id="request-1", + trace_id="trace-1", + account_id=USER_ID, + active_workspace_id=TENANT_ID, + ) + + @pytest.fixture def database_app() -> Iterator[Flask]: app = Flask(__name__) @@ -690,48 +699,162 @@ class TestWorkflowStatisticEndpoints: assert query.end is None def test_workflow_daily_runs_statistic(self, database_app: Flask, monkeypatch: pytest.MonkeyPatch) -> None: - repository = DifyAPISQLAlchemyWorkflowRunRepository( - session_maker=sessionmaker(bind=db.engine, expire_on_commit=False) - ) - monkeypatch.setattr( - repository, - "get_daily_runs_statistics", - lambda **_kwargs: [{"date": "2024-01-01"}], - ) - monkeypatch.setattr( - workflow_statistic_module.DifyAPIRepositoryFactory, - "create_api_workflow_run_repository", - lambda *_args, **_kwargs: repository, - ) + service = MagicMock() + service.get_daily_runs.return_value = [{"date": "2024-01-01", "runs": 2}] monkeypatch.setattr( workflow_statistic_module, - "parse_time_range", - lambda *_args, **_kwargs: (None, None), + "application_services", + lambda: SimpleNamespace(workflow_statistics=service), ) + self._patch_statistic_time_range(monkeypatch) api = workflow_statistic_module.WorkflowDailyRunsStatistic() method = unwrap(api.get) + request_context = _make_request_context() with database_app.test_request_context("/"): - account = _make_account() - account.timezone = "UTC" - response = method(api, WorkflowStatisticQuery(), account, app_model=_make_app("app-1", tenant_id="t1")) + response = method( + api, + WorkflowStatisticQuery(), + request_context, + app_model=_make_app("app-1", tenant_id="t1"), + ) - assert response.get_json() == {"data": [{"date": "2024-01-01"}]} + assert response == {"data": [{"date": "2024-01-01", "runs": 2}]} + service.get_daily_runs.assert_called_once_with( + request_context, + app_id="app-1", + start_date=None, + end_date=None, + timezone="UTC", + ) def test_workflow_daily_terminals_statistic(self, database_app: Flask, monkeypatch: pytest.MonkeyPatch) -> None: - repository = DifyAPISQLAlchemyWorkflowRunRepository( - session_maker=sessionmaker(bind=db.engine, expire_on_commit=False) + service = MagicMock() + service.get_daily_terminals.return_value = [{"date": "2024-01-02", "terminal_count": 3}] + monkeypatch.setattr( + workflow_statistic_module, + "application_services", + lambda: SimpleNamespace(workflow_statistics=service), + ) + self._patch_statistic_time_range(monkeypatch) + + api = workflow_statistic_module.WorkflowDailyTerminalsStatistic() + method = unwrap(api.get) + request_context = _make_request_context() + + with database_app.test_request_context("/"): + response = method( + api, + WorkflowStatisticQuery(), + request_context, + app_model=_make_app("app-1", tenant_id="t1"), + ) + + assert response == {"data": [{"date": "2024-01-02", "terminal_count": 3}]} + service.get_daily_terminals.assert_called_once_with( + request_context, + app_id="app-1", + start_date=None, + end_date=None, + timezone="UTC", + ) + + def test_workflow_daily_token_cost_statistic( + self, + database_app: Flask, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + service = MagicMock() + service.get_daily_token_costs.return_value = [{"date": "2024-01-03", "token_count": 4}] + monkeypatch.setattr( + workflow_statistic_module, + "application_services", + lambda: SimpleNamespace(workflow_statistics=service), + ) + self._patch_statistic_time_range(monkeypatch) + + api = workflow_statistic_module.WorkflowDailyTokenCostStatistic() + method = unwrap(api.get) + request_context = _make_request_context() + + with database_app.test_request_context("/"): + response = method( + api, + WorkflowStatisticQuery(), + request_context, + app_model=_make_app("app-1", tenant_id="t1"), + ) + + assert response == {"data": [{"date": "2024-01-03", "token_count": 4}]} + service.get_daily_token_costs.assert_called_once_with( + request_context, + app_id="app-1", + start_date=None, + end_date=None, + timezone="UTC", + ) + + def test_workflow_average_app_interaction_statistic( + self, + database_app: Flask, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + service = MagicMock() + service.get_average_app_interactions.return_value = [{"date": "2024-01-04", "interactions": 2.5}] + monkeypatch.setattr( + workflow_statistic_module, + "application_services", + lambda: SimpleNamespace(workflow_statistics=service), + ) + self._patch_statistic_time_range(monkeypatch) + + api = workflow_statistic_module.WorkflowAverageAppInteractionStatistic() + method = unwrap(api.get) + request_context = _make_request_context() + + with database_app.test_request_context("/"): + response = method( + api, + WorkflowStatisticQuery(), + request_context, + app_model=_make_app("app-1", tenant_id="t1"), + ) + + assert response == {"data": [{"date": "2024-01-04", "interactions": 2.5}]} + service.get_average_app_interactions.assert_called_once_with( + request_context, + app_id="app-1", + start_date=None, + end_date=None, + timezone="UTC", + ) + + def test_workflow_statistic_invalid_time_range(self, monkeypatch: pytest.MonkeyPatch) -> None: + account = _make_account() + account.timezone = "UTC" + monkeypatch.setattr( + workflow_statistic_module, + "current_account_with_tenant", + lambda: SimpleNamespace(account=account), ) monkeypatch.setattr( - repository, - "get_daily_terminals_statistics", - lambda **_kwargs: [{"date": "2024-01-02"}], + workflow_statistic_module, + "parse_time_range", + MagicMock(side_effect=ValueError("invalid range")), ) + + with pytest.raises(BadRequest, match="invalid range"): + workflow_statistic_module._resolve_statistic_time_range(WorkflowStatisticQuery()) + + @staticmethod + def _patch_statistic_time_range(monkeypatch: pytest.MonkeyPatch) -> None: + account = _make_account() + account.timezone = "UTC" monkeypatch.setattr( - workflow_statistic_module.DifyAPIRepositoryFactory, - "create_api_workflow_run_repository", - lambda *_args, **_kwargs: repository, + workflow_statistic_module, + "current_account_with_tenant", + lambda: SimpleNamespace(account=account), ) monkeypatch.setattr( workflow_statistic_module, @@ -739,16 +862,6 @@ class TestWorkflowStatisticEndpoints: lambda *_args, **_kwargs: (None, None), ) - api = workflow_statistic_module.WorkflowDailyTerminalsStatistic() - method = unwrap(api.get) - - with database_app.test_request_context("/"): - account = _make_account() - account.timezone = "UTC" - response = method(api, WorkflowStatisticQuery(), account, app_model=_make_app("app-1", tenant_id="t1")) - - assert response.get_json() == {"data": [{"date": "2024-01-02"}]} - class TestWorkflowTriggerEndpoints: def test_webhook_trigger_payload(self): diff --git a/api/tests/unit_tests/extensions/test_ext_application_services.py b/api/tests/unit_tests/extensions/test_ext_application_services.py index 78a58828481..f53ba6db671 100644 --- a/api/tests/unit_tests/extensions/test_ext_application_services.py +++ b/api/tests/unit_tests/extensions/test_ext_application_services.py @@ -41,6 +41,7 @@ from services.init_validation_service import InvalidInitializationPasswordError from services.partner_tenant_binding_service import PartnerTenantBindingService from services.tag_application_service import TagApplicationService from services.webapp_access_query_service import WebAppAccessUnavailableError +from services.workflow_statistic_query_service import WorkflowStatisticQueryService @pytest.mark.parametrize( @@ -111,6 +112,7 @@ def test_init_app_registers_services_for_the_current_app( services = ext_application_services.application_services() assert services is app.extensions["application_services"] assert services.init_validation.is_validated(session_validated=False) is False + assert isinstance(services.workflow_statistics, WorkflowStatisticQueryService) @pytest.mark.parametrize( diff --git a/api/tests/unit_tests/services/test_workflow_statistic_query_service.py b/api/tests/unit_tests/services/test_workflow_statistic_query_service.py new file mode 100644 index 00000000000..fde9b86b739 --- /dev/null +++ b/api/tests/unit_tests/services/test_workflow_statistic_query_service.py @@ -0,0 +1,88 @@ +from datetime import UTC, datetime +from unittest.mock import MagicMock + +import pytest + +from machinery.context import RequestContext +from models.enums import WorkflowRunTriggeredFrom +from repositories.api_workflow_run_repository import APIWorkflowRunRepository +from services.workflow_statistic_query_service import WorkflowStatisticQueryService + + +def _request_context(*, workspace_id: str | None = "workspace-1") -> RequestContext: + return RequestContext( + request_id="request-1", + trace_id="trace-1", + account_id="account-1", + active_workspace_id=workspace_id, + ) + + +def test_workflow_statistic_queries_delegate_to_workflow_run_repository() -> None: + workflow_runs = MagicMock(spec=APIWorkflowRunRepository) + workflow_runs.get_daily_runs_statistics.return_value = [{"date": "2024-01-01", "runs": 2}] + workflow_runs.get_daily_terminals_statistics.return_value = [{"date": "2024-01-01", "terminal_count": 3}] + workflow_runs.get_daily_token_cost_statistics.return_value = [{"date": "2024-01-01", "token_count": 4}] + workflow_runs.get_average_app_interaction_statistics.return_value = [{"date": "2024-01-01", "interactions": 2.5}] + service = WorkflowStatisticQueryService(workflow_runs=workflow_runs) + context = _request_context() + start_date = datetime(2024, 1, 1, tzinfo=UTC) + end_date = datetime(2024, 1, 2, tzinfo=UTC) + + assert service.get_daily_runs( + context, + app_id="app-1", + start_date=start_date, + end_date=end_date, + timezone="Asia/Shanghai", + ) == [{"date": "2024-01-01", "runs": 2}] + assert service.get_daily_terminals( + context, + app_id="app-1", + start_date=start_date, + end_date=end_date, + timezone="Asia/Shanghai", + ) == [{"date": "2024-01-01", "terminal_count": 3}] + assert service.get_daily_token_costs( + context, + app_id="app-1", + start_date=start_date, + end_date=end_date, + timezone="Asia/Shanghai", + ) == [{"date": "2024-01-01", "token_count": 4}] + assert service.get_average_app_interactions( + context, + app_id="app-1", + start_date=start_date, + end_date=end_date, + timezone="Asia/Shanghai", + ) == [{"date": "2024-01-01", "interactions": 2.5}] + + expected_arguments = { + "tenant_id": "workspace-1", + "app_id": "app-1", + "triggered_from": WorkflowRunTriggeredFrom.APP_RUN, + "start_date": start_date, + "end_date": end_date, + "timezone": "Asia/Shanghai", + } + workflow_runs.get_daily_runs_statistics.assert_called_once_with(**expected_arguments) + workflow_runs.get_daily_terminals_statistics.assert_called_once_with(**expected_arguments) + workflow_runs.get_daily_token_cost_statistics.assert_called_once_with(**expected_arguments) + workflow_runs.get_average_app_interaction_statistics.assert_called_once_with(**expected_arguments) + + +def test_workflow_statistic_query_requires_active_workspace() -> None: + workflow_runs = MagicMock(spec=APIWorkflowRunRepository) + service = WorkflowStatisticQueryService(workflow_runs=workflow_runs) + + with pytest.raises(RuntimeError, match="did not resolve an active workspace"): + service.get_daily_runs( + _request_context(workspace_id=None), + app_id="app-1", + start_date=None, + end_date=None, + timezone="UTC", + ) + + workflow_runs.get_daily_runs_statistics.assert_not_called()