Skip to content

Data Transfer Objects (DTOs)

Application Layer - DTOs Package.

This package contains Data Transfer Objects (DTOs) used for transferring data between different layers of the application while maintaining loose coupling.

Modules

Package Overview

Data Transfer Objects (DTOs) Module.

This module contains immutable data transfer objects that encapsulate data crossing layer boundaries. DTOs prevent tight coupling between layers and provide clear contracts for data exchange.

Classes:

Name Description
UploadResultDTO

Encapsulates upload operation results with validation status
ValidationResultDTO

Contains validation results with detailed error messages
MergedDataDTO

Represents merged data results from database integration
ProcessingProgressDTO

Encapsulates processing progress information
Notes

All DTOs are immutable frozen dataclasses to ensure data integrity across layer boundaries. They contain only data, no business logic.

UploadResultDTO

UploadResultDTO dataclass 

UploadResultDTO(success: bool, dataset: Optional[Dataset], filename: str, sample_count: int, ko_count: int, message: str, errors: Optional[list[str]] = None)

Immutable DTO for upload operation results.

Encapsulates all relevant information from a file upload operation, including success status, parsed dataset, and metadata.

Parameters:

Name Type Description Default
success bool

Whether the upload operation succeeded

 required
dataset Optional[Dataset]

The parsed dataset entity (None if upload failed)

 required
filename str

Name of the uploaded file

 required
sample_count int

Number of samples parsed

 required
ko_count int

Total number of KOs across all samples

 required
message str

User-friendly message about the upload result

 required
errors Optional[list[str]]

List of validation or parsing errors (None if no errors)

 None

Attributes:

Name Type Description
success bool

Upload success status
dataset Optional[Dataset]

Parsed dataset or None
filename str

Uploaded filename
sample_count int

Sample count
ko_count int

KO count
message str

Result message
errors Optional[list[str]]

Error messages if any
Notes
  • This DTO is immutable (frozen) to ensure data integrity
  • All validation should occur before DTO creation

Functions

__post_init__ 
__post_init__() -> None

Validate DTO consistency after initialization.

Raises:

Type Description
ValueError

If success=True but dataset is None, or if counts are negative
Notes

This method is called automatically by dataclass after init. It ensures logical consistency of the DTO state.

Source code in src/application/dto/upload_result_dto.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __post_init__(self) -> None:
    """
    Validate DTO consistency after initialization.

    Raises
    ------
    ValueError
        If success=True but dataset is None, or if counts are negative

    Notes
    -----
    This method is called automatically by dataclass after __init__.
    It ensures logical consistency of the DTO state.
    """
    if self.success and self.dataset is None:
        raise ValueError("Success=True requires a valid dataset")

    if self.sample_count < 0 or self.ko_count < 0:
        raise ValueError("Counts cannot be negative")
has_errors 
has_errors() -> bool

Check if upload had any errors.

Returns:

Type Description
bool

True if errors list is not empty, False otherwise
Source code in src/application/dto/upload_result_dto.py 
 96
 97
 98
 99
100
101
102
103
104
105
def has_errors(self) -> bool:
    """
    Check if upload had any errors.

    Returns
    -------
    bool
        True if errors list is not empty, False otherwise
    """
    return self.errors is not None and len(self.errors) > 0

MergedDataDTO

MergedDataDTO dataclass 

MergedDataDTO(biorempp_data: DataFrame, hadeg_data: Optional[DataFrame], toxcsm_data: Optional[DataFrame], match_count: int, total_records: int, cache_key: str, processing_time_seconds: float = 0.0)

Immutable DTO for database merge operation results.

Encapsulates merged data from BioRemPP, HADEG, and ToxCSM databases along with metadata about the merge operation.

Parameters:

Name Type Description Default
biorempp_data DataFrame

Data merged with BioRemPP database

 required
hadeg_data Optional[DataFrame]

Data merged with HADEG database (None if not merged)

 required
toxcsm_data Optional[DataFrame]

Data merged with ToxCSM database (None if not merged)

 required
match_count int

Number of successful matches across all databases

 required
total_records int

Total number of input records processed

 required
cache_key str

Cache key for storing/retrieving this merge result

 required
processing_time_seconds float

Time taken for merge operation in seconds

 0.0

Attributes:

Name Type Description
biorempp_data DataFrame

BioRemPP merged data
hadeg_data Optional[DataFrame]

HADEG merged data
toxcsm_data Optional[DataFrame]

ToxCSM merged data
match_count int

Match count
total_records int

Total records
cache_key str

Cache key
processing_time_seconds float

Processing time
Notes
  • This DTO is immutable (frozen)
  • DataFrames should be copied before modification to maintain immutability

Functions

__post_init__ 
__post_init__() -> None

Validate DTO consistency.

Raises:

Type Description
ValueError

If counts are negative or match_count > total_records
TypeError

If biorempp_data is not a DataFrame
Source code in src/application/dto/merged_data_dto.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def __post_init__(self) -> None:
    """
    Validate DTO consistency.

    Raises
    ------
    ValueError
        If counts are negative or match_count > total_records
    TypeError
        If biorempp_data is not a DataFrame
    """
    if not isinstance(self.biorempp_data, pd.DataFrame):
        raise TypeError("biorempp_data must be a pandas DataFrame")

    if self.match_count < 0 or self.total_records < 0:
        raise ValueError("Counts cannot be negative")

    if self.match_count > self.total_records:
        raise ValueError("match_count cannot exceed total_records")

    if self.processing_time_seconds < 0:
        raise ValueError("processing_time_seconds cannot be negative")
match_rate 
match_rate() -> float

Calculate match rate percentage.

Returns:

Type Description
float

Percentage of records matched (0.0 to 100.0)
Source code in src/application/dto/merged_data_dto.py 
 99
100
101
102
103
104
105
106
107
108
109
110
def match_rate(self) -> float:
    """
    Calculate match rate percentage.

    Returns
    -------
    float
        Percentage of records matched (0.0 to 100.0)
    """
    if self.total_records == 0:
        return 0.0
    return (self.match_count / self.total_records) * 100.0
has_hadeg_data 
has_hadeg_data() -> bool

Check if HADEG data is available.

Returns:

Type Description
bool

True if HADEG data exists
Source code in src/application/dto/merged_data_dto.py 
112
113
114
115
116
117
118
119
120
121
def has_hadeg_data(self) -> bool:
    """
    Check if HADEG data is available.

    Returns
    -------
    bool
        True if HADEG data exists
    """
    return self.hadeg_data is not None and not self.hadeg_data.empty
has_toxcsm_data 
has_toxcsm_data() -> bool

Check if ToxCSM data is available.

Returns:

Type Description
bool

True if ToxCSM data exists
Source code in src/application/dto/merged_data_dto.py 
123
124
125
126
127
128
129
130
131
132
def has_toxcsm_data(self) -> bool:
    """
    Check if ToxCSM data is available.

    Returns
    -------
    bool
        True if ToxCSM data exists
    """
    return self.toxcsm_data is not None and not self.toxcsm_data.empty

ProcessingProgressDTO

ProcessingProgressDTO dataclass 

ProcessingProgressDTO(current_stage: str, stage_number: int, total_stages: int, progress_percentage: float, message: str = '', estimated_time_remaining: Optional[float] = None, error: Optional[str] = None)

Deprecated Immutable DTO for processing progress information.

Functions

__post_init__ 
__post_init__() -> None

Validate DTO consistency.

Raises:

Type Description
ValueError

If stage numbers invalid or percentage out of range
Source code in src/application/dto/processing_progress_dto.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __post_init__(self) -> None:
    """
    Validate DTO consistency.

    Raises
    ------
    ValueError
        If stage numbers invalid or percentage out of range
    """
    if self.total_stages < 1:
        raise ValueError("total_stages must be at least 1")

    if self.stage_number < 1 or self.stage_number > self.total_stages:
        raise ValueError(f"stage_number must be between 1 and {self.total_stages}")

    if not 0.0 <= self.progress_percentage <= 100.0:
        raise ValueError("progress_percentage must be between 0.0 and 100.0")

    if (
        self.estimated_time_remaining is not None
        and self.estimated_time_remaining < 0
    ):
        raise ValueError("estimated_time_remaining cannot be negative")
is_complete 
is_complete() -> bool

Check if processing is complete.

Returns:

Type Description
bool

True if progress is 100%
Source code in src/application/dto/processing_progress_dto.py 
56
57
58
59
60
61
62
63
64
65
def is_complete(self) -> bool:
    """
    Check if processing is complete.

    Returns
    -------
    bool
        True if progress is 100%
    """
    return self.progress_percentage >= 100.0
has_error 
has_error() -> bool

Check if processing has error.

Returns:

Type Description
bool

True if error exists
Source code in src/application/dto/processing_progress_dto.py 
67
68
69
70
71
72
73
74
75
76
def has_error(self) -> bool:
    """
    Check if processing has error.

    Returns
    -------
    bool
        True if error exists
    """
    return self.error is not None
is_final_stage 
is_final_stage() -> bool

Check if currently on final stage.

Returns:

Type Description
bool

True if on last stage
Source code in src/application/dto/processing_progress_dto.py 
78
79
80
81
82
83
84
85
86
87
def is_final_stage(self) -> bool:
    """
    Check if currently on final stage.

    Returns
    -------
    bool
        True if on last stage
    """
    return self.stage_number == self.total_stages

ValidationResultDTO

ValidationResultDTO dataclass 

ValidationResultDTO(is_valid: bool, errors: Optional[list[str]], warnings: Optional[list[str]], validated_items: int, message: str = '')

Immutable DTO for validation operation results.

Encapsulates validation outcomes including success status, error messages, warnings, and metadata about validated items.

Parameters:

Name Type Description Default
is_valid bool

Whether validation passed (no errors)

 required
errors Optional[list[str]]

List of validation error messages (None if no errors)

 required
warnings Optional[list[str]]

List of validation warnings (None if no warnings)

 required
validated_items int

Number of items validated

 required
message str

Summary message about validation result

 ""

Attributes:

Name Type Description
is_valid bool

Validation success status
errors Optional[list[str]]

Error messages
warnings Optional[list[str]]

Warning messages
validated_items int

Count of validated items
message str

Summary message
Notes
  • This DTO is immutable (frozen) to ensure data integrity
  • Errors prevent processing, warnings allow processing but flag issues

Functions

__post_init__ 
__post_init__() -> None

Validate DTO consistency.

Raises:

Type Description
ValueError

If is_valid=False but no errors, or validated_items is negative
Source code in src/application/dto/validation_result_dto.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __post_init__(self) -> None:
    """
    Validate DTO consistency.

    Raises
    ------
    ValueError
        If is_valid=False but no errors, or validated_items is negative
    """
    if not self.is_valid and (self.errors is None or len(self.errors) == 0):
        raise ValueError("is_valid=False requires error messages")

    if self.validated_items < 0:
        raise ValueError("validated_items cannot be negative")
has_errors 
has_errors() -> bool

Check if validation has errors.

Returns:

Type Description
bool

True if errors exist
Source code in src/application/dto/validation_result_dto.py 
79
80
81
82
83
84
85
86
87
88
def has_errors(self) -> bool:
    """
    Check if validation has errors.

    Returns
    -------
    bool
        True if errors exist
    """
    return self.errors is not None and len(self.errors) > 0
has_warnings 
has_warnings() -> bool

Check if validation has warnings.

Returns:

Type Description
bool

True if warnings exist
Source code in src/application/dto/validation_result_dto.py 
90
91
92
93
94
95
96
97
98
99
def has_warnings(self) -> bool:
    """
    Check if validation has warnings.

    Returns
    -------
    bool
        True if warnings exist
    """
    return self.warnings is not None and len(self.warnings) > 0
error_count 
error_count() -> int

Get count of errors.

Returns:

Type Description
int

Number of errors
Source code in src/application/dto/validation_result_dto.py 
101
102
103
104
105
106
107
108
109
110
def error_count(self) -> int:
    """
    Get count of errors.

    Returns
    -------
    int
        Number of errors
    """
    return len(self.errors) if self.errors else 0
warning_count 
warning_count() -> int

Get count of warnings.

Returns:

Type Description
int

Number of warnings
Source code in src/application/dto/validation_result_dto.py 
112
113
114
115
116
117
118
119
120
121
def warning_count(self) -> int:
    """
    Get count of warnings.

    Returns
    -------
    int
        Number of warnings
    """
    return len(self.warnings) if self.warnings else 0