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 0x0000021FE0F5D3F0>

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

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

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

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

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

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

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

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

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

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

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

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

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 0x0000021FE0EF3190>

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

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

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

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

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

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

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

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

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

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

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

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

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 0x0000021FE0DFB250>

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

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

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

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

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

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

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

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

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

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

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 0x0000021FE18F1000>

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

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

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

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

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

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

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

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 0x0000021FE0DA00A0>

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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 0x0000021FE0D82500>

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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.