Metrics

dotflow.abc.metrics.Metrics

Bases: ABC

Metrics

Source code in dotflow/abc/metrics.py

class Metrics(ABC):
    """Metrics"""

    @abstractmethod
    def workflow_started(self, workflow_id: Any, **kwargs) -> None:
        """Called when a workflow starts."""

    @abstractmethod
    def workflow_completed(self, workflow_id: Any, duration: float) -> None:
        """Called when a workflow completes successfully."""

    @abstractmethod
    def workflow_failed(self, workflow_id: Any, duration: float) -> None:
        """Called when a workflow fails."""

    @abstractmethod
    def task_completed(self, task: Any) -> None:
        """Called when a task completes successfully."""

    @abstractmethod
    def task_failed(self, task: Any) -> None:
        """Called when a task fails."""

    @abstractmethod
    def task_retried(self, task: Any) -> None:
        """Called when a task is retried."""

task_completed(task) abstractmethod

Called when a task completes successfully.

Source code in dotflow/abc/metrics.py

@abstractmethod
def task_completed(self, task: Any) -> None:
    """Called when a task completes successfully."""

task_failed(task) abstractmethod

Called when a task fails.

Source code in dotflow/abc/metrics.py

@abstractmethod
def task_failed(self, task: Any) -> None:
    """Called when a task fails."""

task_retried(task) abstractmethod

Called when a task is retried.

Source code in dotflow/abc/metrics.py

@abstractmethod
def task_retried(self, task: Any) -> None:
    """Called when a task is retried."""

workflow_completed(workflow_id, duration) abstractmethod

Called when a workflow completes successfully.

Source code in dotflow/abc/metrics.py

@abstractmethod
def workflow_completed(self, workflow_id: Any, duration: float) -> None:
    """Called when a workflow completes successfully."""

workflow_failed(workflow_id, duration) abstractmethod

Called when a workflow fails.

Source code in dotflow/abc/metrics.py

@abstractmethod
def workflow_failed(self, workflow_id: Any, duration: float) -> None:
    """Called when a workflow fails."""

workflow_started(workflow_id, **kwargs) abstractmethod

Called when a workflow starts.

Source code in dotflow/abc/metrics.py

@abstractmethod
def workflow_started(self, workflow_id: Any, **kwargs) -> None:
    """Called when a workflow starts."""