Skip to content

Analysis

analysis

Analysis Entity

Represents metadata of an analysis executed in the system.

Classes

AnalysisStatus

Bases: Enum

Possible statuses of an analysis.

Attributes:

Name Type Description
PENDING str

Analysis awaiting execution.

RUNNING str

Analysis in progress.

COMPLETED str

Analysis completed successfully.

FAILED str

Analysis failed.

CACHED str

Result loaded from cache.

Analysis dataclass

Analysis(id: str, name: str, category: str, status: AnalysisStatus = AnalysisStatus.PENDING, created_at: datetime = datetime.now(), started_at: Optional[datetime] = None, completed_at: Optional[datetime] = None, config: Dict[str, Any] = dict(), result_metadata: Dict[str, Any] = dict(), error_message: Optional[str] = None)

Entity for analysis metadata.

Stores information about an analysis that has been or will be executed, including identification, status, timestamps, and configurations.

Parameters:

Name Type Description Default
id str

Unique identifier for the analysis (e.g., 'UC1_1', 'UC2_3')

required
name str

Descriptive name of the analysis

required
category str

Category of the analysis (e.g., 'heatmaps', 'rankings')

required
status AnalysisStatus

Current status of the analysis

PENDING
created_at datetime

Creation timestamp

datetime.now()
started_at Optional[datetime]

Execution start timestamp

None
completed_at Optional[datetime]

Completion timestamp

None
config Dict[str, Any]

Specific configurations for the analysis

{}
result_metadata Dict[str, Any]

Result metadata

{}
error_message Optional[str]

Error message if it failed

None
Notes

This entity is used for tracking and auditing analyses.

Attributes
duration_seconds property
duration_seconds: Optional[float]

Calculates the analysis duration in seconds.

Returns:

Type Description
Optional[float]

Duration in seconds or None if not finished.

is_completed property
is_completed: bool

Checks if the analysis is completed (success or failure).

Returns:

Type Description
bool

True if status is COMPLETED, FAILED, or CACHED.

is_successful property
is_successful: bool

Checks if the analysis was successful.

Returns:

Type Description
bool

True if status is COMPLETED or CACHED.

Functions
start
start() -> None

Marks the analysis as started.

Notes

Updates status to RUNNING and records the start timestamp.

Source code in src/domain/entities/analysis.py
def start(self) -> None:
    """
    Marks the analysis as started.

    Notes
    -----
    Updates status to RUNNING and records the start timestamp.
    """
    self.status = AnalysisStatus.RUNNING
    self.started_at = datetime.now()
complete
complete(metadata: Optional[Dict[str, Any]] = None) -> None

Marks the analysis as successfully completed.

Parameters:

Name Type Description Default
metadata Optional[Dict[str, Any]]

Result metadata, by default None.

None
Notes

Updates status to COMPLETED and records the completion timestamp.

Source code in src/domain/entities/analysis.py
def complete(self, metadata: Optional[Dict[str, Any]] = None) -> None:
    """
    Marks the analysis as successfully completed.

    Parameters
    ----------
    metadata : Optional[Dict[str, Any]], optional
        Result metadata, by default None.

    Notes
    -----
    Updates status to COMPLETED and records the completion timestamp.
    """
    self.status = AnalysisStatus.COMPLETED
    self.completed_at = datetime.now()
    if metadata:
        self.result_metadata.update(metadata)
fail
fail(error_message: str) -> None

Marks the analysis as failed.

Parameters:

Name Type Description Default
error_message str

Error message describing the failure.

required
Notes

Updates status to FAILED and records the error message.

Source code in src/domain/entities/analysis.py
def fail(self, error_message: str) -> None:
    """
    Marks the analysis as failed.

    Parameters
    ----------
    error_message : str
        Error message describing the failure.

    Notes
    -----
    Updates status to FAILED and records the error message.
    """
    self.status = AnalysisStatus.FAILED
    self.completed_at = datetime.now()
    self.error_message = error_message
mark_from_cache
mark_from_cache() -> None

Marks the analysis as loaded from cache.

Notes

The CACHED status indicates that the result was retrieved from the cache without the need for reprocessing.

Source code in src/domain/entities/analysis.py
def mark_from_cache(self) -> None:
    """
    Marks the analysis as loaded from cache.

    Notes
    -----
    The CACHED status indicates that the result was retrieved from
    the cache without the need for reprocessing.
    """
    self.status = AnalysisStatus.CACHED
    self.completed_at = datetime.now()
validate
validate() -> None

Validates the analysis metadata.

Raises:

Type Description
ValueError

If ID, name, or category are empty.

Source code in src/domain/entities/analysis.py
def validate(self) -> None:
    """
    Validates the analysis metadata.

    Raises
    ------
    ValueError
        If ID, name, or category are empty.
    """
    if not self.id or not self.id.strip():
        raise ValueError("Analysis ID cannot be empty")

    if not self.name or not self.name.strip():
        raise ValueError("Analysis name cannot be empty")

    if not self.category or not self.category.strip():
        raise ValueError("Analysis category cannot be empty")
__str__
__str__() -> str

Returns the string representation of the analysis.

Returns:

Type Description
str

Descriptive string.

Source code in src/domain/entities/analysis.py
def __str__(self) -> str:
    """
    Returns the string representation of the analysis.

    Returns
    -------
    str
        Descriptive string.
    """
    return f"Analysis({self.id}: {self.name}) - {self.status.value}"
__repr__
__repr__() -> str

Returns the debug representation of the analysis.

Returns:

Type Description
str

Detailed representation.

Source code in src/domain/entities/analysis.py
def __repr__(self) -> str:
    """
    Returns the debug representation of the analysis.

    Returns
    -------
    str
        Detailed representation.
    """
    return (
        f"Analysis(id='{self.id}', "
        f"name='{self.name}', "
        f"status={self.status})"
    )