""" Service-layer repository protocol for WorkflowNodeExecutionModel operations. This module provides a protocol interface for service-layer operations on WorkflowNodeExecutionModel that abstracts database queries currently done directly in service classes. This repository is specifically designed for service-layer needs and is separate from the core domain repository. The service repository handles operations that require access to database-specific fields like tenant_id, app_id, triggered_from, etc., which are not part of the core domain model. """ from collections.abc import Sequence from datetime import datetime from typing import Optional, Protocol from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository from models.workflow import WorkflowNodeExecutionModel class DifyAPIWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository, Protocol): """ Protocol for service-layer operations on WorkflowNodeExecutionModel. This repository provides database access patterns specifically needed by service classes, handling queries that involve database-specific fields and multi-tenancy concerns. Key responsibilities: - Manages database operations for workflow node executions - Handles multi-tenant data isolation - Provides batch processing capabilities - Supports execution lifecycle management Implementation notes: - Returns database models directly (WorkflowNodeExecutionModel) - Handles tenant/app filtering automatically - Provides service-specific query patterns - Focuses on database operations without domain logic - Supports cleanup and maintenance operations """ def get_node_last_execution( self, tenant_id: str, app_id: str, workflow_id: str, node_id: str, ) -> Optional[WorkflowNodeExecutionModel]: """ Get the most recent execution for a specific node. This method finds the latest execution of a specific node within a workflow, ordered by creation time. Used primarily for debugging and inspection purposes. Args: tenant_id: The tenant identifier app_id: The application identifier workflow_id: The workflow identifier node_id: The node identifier Returns: The most recent WorkflowNodeExecutionModel for the node, or None if not found """ ... def get_executions_by_workflow_run( self, tenant_id: str, app_id: str, workflow_run_id: str, ) -> Sequence[WorkflowNodeExecutionModel]: """ Get all node executions for a specific workflow run. This method retrieves all node executions that belong to a specific workflow run, ordered by index in descending order for proper trace visualization. Args: tenant_id: The tenant identifier app_id: The application identifier workflow_run_id: The workflow run identifier Returns: A sequence of WorkflowNodeExecutionModel instances ordered by index (desc) """ ... def get_execution_by_id( self, execution_id: str, tenant_id: Optional[str] = None, ) -> Optional[WorkflowNodeExecutionModel]: """ Get a workflow node execution by its ID. This method retrieves a specific execution by its unique identifier. Tenant filtering is optional for cases where the execution ID is globally unique. When `tenant_id` is None, it's the caller's responsibility to ensure proper data isolation between tenants. If the `execution_id` comes from untrusted sources (e.g., retrieved from an API request), the caller should set `tenant_id` to prevent horizontal privilege escalation. Args: execution_id: The execution identifier tenant_id: Optional tenant identifier for additional filtering Returns: The WorkflowNodeExecutionModel if found, or None if not found """ ... def delete_expired_executions( self, tenant_id: str, before_date: datetime, batch_size: int = 1000, ) -> int: """ Delete workflow node executions that are older than the specified date. This method is used for cleanup operations to remove expired executions in batches to avoid overwhelming the database. Args: tenant_id: The tenant identifier before_date: Delete executions created before this date batch_size: Maximum number of executions to delete in one batch Returns: The number of executions deleted """ ... def delete_executions_by_app( self, tenant_id: str, app_id: str, batch_size: int = 1000, ) -> int: """ Delete all workflow node executions for a specific app. This method is used when removing an app and all its related data. Executions are deleted in batches to avoid overwhelming the database. Args: tenant_id: The tenant identifier app_id: The application identifier batch_size: Maximum number of executions to delete in one batch Returns: The total number of executions deleted """ ... def get_expired_executions_batch( self, tenant_id: str, before_date: datetime, batch_size: int = 1000, ) -> Sequence[WorkflowNodeExecutionModel]: """ Get a batch of expired workflow node executions for backup purposes. This method retrieves expired executions without deleting them, allowing the caller to backup the data before deletion. Args: tenant_id: The tenant identifier before_date: Get executions created before this date batch_size: Maximum number of executions to retrieve Returns: A sequence of WorkflowNodeExecutionModel instances """ ... def delete_executions_by_ids( self, execution_ids: Sequence[str], ) -> int: """ Delete workflow node executions by their IDs. This method deletes specific executions by their IDs, typically used after backing up the data. This method does not perform tenant isolation checks. The caller is responsible for ensuring proper data isolation between tenants. When execution IDs come from untrusted sources (e.g., API requests), additional tenant validation should be implemented to prevent unauthorized access. Args: execution_ids: List of execution IDs to delete Returns: The number of executions deleted """ ...