dataclass.logs

Data cleaning and pipeline execution logging structures.

This module provides Pydantic models for structured logging of data cleaning operations, designed to be easily consumed by web frontends for beautiful and informative displays.

Classes

class dataclass.logs.LogLevel

Log level enumeration for consistent severity classification.

Inherits from:: str, Enum

Attributes:

DEBUG = 'debug'

INFO = 'info'

WARNING = 'warning'

ERROR = 'error'

CRITICAL = 'critical'

class dataclass.logs.ActionType

Types of actions that can be performed during data cleaning.

Inherits from:: str, Enum

Attributes:

NO_ACTION = 'no_action'

REMOVE_ROWS = 'remove_rows'

REMOVE_COLUMNS = 'remove_columns'

FILL_VALUES = 'fill_values'

CONVERT_TYPES = 'convert_types'

VALIDATE_DATA = 'validate_data'

FILTER_DATA = 'filter_data'

TRANSFORM_DATA = 'transform_data'

class dataclass.logs.IssueType

Types of data quality issues that can be detected.

Inherits from:: str, Enum

Attributes:

MISSING_VALUES = 'missing_values'

INVALID_TYPE = 'invalid_type'

OUT_OF_RANGE = 'out_of_range'

DUPLICATE_VALUES = 'duplicate_values'

INCONSISTENT_FORMAT = 'inconsistent_format'

CONSTRAINT_VIOLATION = 'constraint_violation'

DATA_ANOMALY = 'data_anomaly'

class dataclass.logs.IssueDetail

Detailed information about a specific data quality issue.

Inherits from:: BaseModel

Properties:

affected_percentage: Calculate the percentage of affected records.

severity_score: Convert severity to numeric score for sorting (higher = more severe).

Attributes:

id: UUID4 = <ast.Call object at 0x0000017357D5D3C0>

issue_type: IssueType = <ast.Call object at 0x0000017357D5D210>

severity: LogLevel = <ast.Call object at 0x0000017357D5D090>

table_name: str | None = <ast.Call object at 0x0000017357D5CEB0>

column_name: str | None = <ast.Call object at 0x0000017357D5CCD0>

row_indices: list[int] | None = <ast.Call object at 0x0000017357D5CA90>

description: str = <ast.Call object at 0x0000017357D5C910>

detected_value: Any = <ast.Call object at 0x0000017357D5C790>

expected_value: Any = <ast.Call object at 0x0000017357D5C610>

affected_count: int = <ast.Call object at 0x0000017357D5C490>

total_count: int = <ast.Call object at 0x0000017357D5C310>

rule_name: str | None = <ast.Call object at 0x0000017357D5C130>

context: dict[(str, Any)] = <ast.Call object at 0x0000017357D1FEB0>

class dataclass.logs.ActionDetail

Detailed information about an action taken during data cleaning.

Inherits from:: BaseModel

Properties:

records_changed: Calculate the number of records changed.

columns_changed: Calculate the number of columns changed.

Attributes:

id: UUID4 = <ast.Call object at 0x0000017357D1F160>

action_type: ActionType = <ast.Call object at 0x0000017357D1EFB0>

timestamp: datetime = <ast.Call object at 0x0000017357D1EE30>

description: str = <ast.Call object at 0x0000017357D1EC50>

table_name: str | None = <ast.Call object at 0x0000017357D1EA70>

column_names: list[str] | None = <ast.Call object at 0x0000017357D1E830>

records_before: int = <ast.Call object at 0x0000017357D1E6B0>

records_after: int = <ast.Call object at 0x0000017357D01780>

columns_before: int = <ast.Call object at 0x0000017357D01C60>

columns_after: int = <ast.Call object at 0x0000017357D01990>

parameters: dict[(str, Any)] = <ast.Call object at 0x0000017357D01C30>

success: bool = <ast.Call object at 0x0000017357D02020>

error_message: str | None = <ast.Call object at 0x0000017357D01E70>

class dataclass.logs.RuleExecutionResult

Result of executing a single cleaning rule.

Inherits from:: BaseModel

Properties:

total_issues: Total number of issues found by this rule.

total_actions: Total number of actions taken by this rule.

critical_issues: Get only critical and error-level issues.

Attributes:

rule_name: str = <ast.Call object at 0x0000017357D03250>

rule_description: str | None = <ast.Call object at 0x0000017357D02F80>

execution_time: datetime = <ast.Call object at 0x0000017357D027A0>

duration_ms: float | None = <ast.Call object at 0x0000017357D029E0>

status: Literal[(success, failed, skipped)] = <ast.Call object at 0x0000017357D02DD0>

enabled: bool = <ast.Call object at 0x0000017357D02D10>

error_message: str | None = <ast.Call object at 0x0000017357D03490>

issues_found: list[IssueDetail] = <ast.Call object at 0x0000017357D03670>

actions_taken: list[ActionDetail] = <ast.Call object at 0x000001735726E110>

records_processed: int = <ast.Call object at 0x000001735726E0E0>

tables_processed: list[str] = <ast.Call object at 0x000001735726C550>

class dataclass.logs.TableProcessingSummary

Summary of processing for a single table.

Inherits from:: BaseModel

Properties:

rows_changed: Number of rows changed.

columns_changed: Number of columns changed.

change_percentage: Percentage of data changed.

Attributes:

table_name: str = <ast.Call object at 0x000001735726C2B0>

original_shape: tuple[(int, int)] = <ast.Call object at 0x000001735726F1F0>

final_shape: tuple[(int, int)] = <ast.Call object at 0x000001735726EC80>

total_issues: int = <ast.Call object at 0x000001735726F970>

issues_by_type: dict[(IssueType, int)] = <ast.Call object at 0x000001735726F5B0>

actions_performed: list[ActionType] = <ast.Call object at 0x000001735726F550>

data_quality_score: float | None = <ast.Call object at 0x000001735726F160>

completeness_score: float | None = <ast.Call object at 0x000001735726EF80>

class dataclass.logs.ModuleExecutionLog

Log entry for a single module execution.

Inherits from:: BaseModel

Properties:

duration_seconds: Calculate execution duration in seconds.

success_rate: Calculate the success rate of rule executions.

critical_issues: Get all critical issues from all rules.

Attributes:

id: UUID4 = <ast.Call object at 0x0000017357238070>

module_name: str = <ast.Call object at 0x0000017357238430>

module_class: str = <ast.Call object at 0x000001735723A110>

start_time: datetime = <ast.Call object at 0x000001735723A5F0>

end_time: datetime | None = <ast.Call object at 0x0000017357238640>

status: Literal[(running, completed, failed, skipped)] = <ast.Call object at 0x0000017357238DC0>

configuration: dict[(str, Any)] = <ast.Call object at 0x0000017357238250>

input_data_info: dict[(str, Any)] = <ast.Call object at 0x000001735723AAD0>

rule_results: list[RuleExecutionResult] = <ast.Call object at 0x000001735723A830>

table_summaries: list[TableProcessingSummary] = <ast.Call object at 0x0000017357239450>

total_records_processed: int = <ast.Call object at 0x0000017357238B50>

total_issues_found: int = <ast.Call object at 0x0000017357239930>

total_actions_taken: int = <ast.Call object at 0x0000017357238790>

error_message: str | None = <ast.Call object at 0x000001735721C5B0>

stack_trace: str | None = <ast.Call object at 0x000001735721D1E0>

class dataclass.logs.PipelineExecutionReport

Complete report for a pipeline execution with data cleaning operations.

Inherits from:: BaseModel

Methods:

get_module_log(module_name: str) → ModuleExecutionLog | None: Get log for a specific module.

get_issues_for_table(table_name: str) → list[IssueDetail]: Get all issues for a specific table.

generate_frontend_summary() → dict[str, Any]: Generate a summary optimized for frontend dashboard display.

Properties:

duration_seconds: Calculate total execution duration in seconds.

success_rate: Calculate overall module success rate.

all_critical_issues: Get all critical issues from all modules.

issues_by_severity: Count issues by severity level.

processing_speed: Calculate records processed per second.

Attributes:

id: UUID4 = <ast.Call object at 0x000001735721F100>

pipeline_name: str = <ast.Call object at 0x000001735721EE00>

pipeline_title: str | None = <ast.Call object at 0x000001735721C430>

start_time: datetime = <ast.Call object at 0x000001735721C4C0>

end_time: datetime | None = <ast.Call object at 0x000001735721E0B0>

status: Literal[(running, completed, failed, cancelled)] = <ast.Call object at 0x000001735721C850>

module_logs: list[ModuleExecutionLog] = <ast.Call object at 0x000001735721C0A0>

total_modules: int = <ast.Call object at 0x000001735721E8C0>

successful_modules: int = <ast.Call object at 0x000001735721F1C0>

total_records_processed: int = <ast.Call object at 0x000001735721F130>

total_issues_found: int = <ast.Call object at 0x000001735721E050>

total_actions_taken: int = <ast.Call object at 0x000001735721C2B0>

overall_quality_score: float | None = <ast.Call object at 0x000001735721F760>

quality_improvement: float | None = <ast.Call object at 0x000001735721CFA0>

pipeline_config: dict[(str, Any)] = <ast.Call object at 0x000001735721FCD0>

environment_info: dict[(str, Any)] = <ast.Call object at 0x000001735721CF70>

summary: dict[(str, Any)] = <ast.Call object at 0x000001735721F1F0>

Functions

dataclass.logs.create_pipeline_execution_report(pipeline_name: str, pipeline_title: str | None = None, pipeline_config: dict[str, Any] | None = None) → PipelineExecutionReport: Create a new pipeline execution report.