Event and Alert Hook System - Implementation Summary

Overview

A comprehensive event emission and alert hook system has been successfully implemented for Baselinr, following the specifications in implement_event_hooks.md. This system enables runtime events to be emitted during profiling and drift detection, processed by multiple registered hooks, and optionally persisted or alerted.

What Was Implemented

1. Core Event System (✅ Completed)

New Modules:

baselinr/events/__init__.py - Package exports
baselinr/events/events.py - Event dataclasses
baselinr/events/hooks.py - AlertHook protocol
baselinr/events/event_bus.py - EventBus implementation
baselinr/events/builtin_hooks.py - Built-in hook implementations

Event Types:

BaseEvent - Base class for all events
DataDriftDetected - Emitted when data drift is detected
SchemaChangeDetected - Emitted when schema changes are detected
ProfilingStarted - Emitted when profiling begins
ProfilingCompleted - Emitted when profiling completes successfully
ProfilingFailed - Emitted when profiling fails

Built-in Hooks:

LoggingAlertHook - Logs events to stdout (useful for development/debugging)
SQLEventHook - Persists events to any SQL database (Postgres, MySQL, SQLite)
SnowflakeEventHook - Persists events to Snowflake with VARIANT support for metadata

2. Integration with Core Components (✅ Completed)

Drift Detector (baselinr/drift/detector.py):

Accepts optional EventBus in constructor
Emits DataDriftDetected events when drift is found
Emits SchemaChangeDetected events for schema changes
Events include table, column, metric, baseline/current values, and severity

Profiling Core (baselinr/profiling/core.py):

Accepts optional EventBus in constructor
Emits ProfilingStarted events when profiling begins
Emits ProfilingCompleted events on successful completion (with duration, row/column counts)
Emits ProfilingFailed events when errors occur

3. Configuration System (✅ Completed)

Schema Updates (baselinr/config/schema.py):

Added HookConfig - Configuration for individual hooks
Added HooksConfig - Master configuration for all hooks
Integrated into BaselinrConfig with hooks field
Support for logging, sql, snowflake, and custom hook types

Configuration Features:

Master enabled switch to toggle all hooks
Per-hook enabled flag for selective activation
Type-specific parameters (log_level, connection, table_name, etc.)
Custom hook support with dynamic module/class loading

4. CLI Integration (✅ Completed)

Updates to baselinr/cli.py:

Added create_event_bus() function to initialize EventBus from config
Added _create_hook() factory function for hook instantiation
Updated profile_command to create and pass EventBus to ProfileEngine
Updated drift_command to create and pass EventBus to DriftDetector
Support for custom hook loading via importlib

5. Database Schema (✅ Completed)

SQL Schema (baselinr/storage/schema.sql):

Added baselinr_events table for event persistence
Fields: event_id, event_type, table_name, column_name, metric_name, baseline_value, current_value, change_percent, drift_severity, timestamp, metadata, created_at
Indexes on event_type, table_name, timestamp, drift_severity

Snowflake Schema (baselinr/storage/schema_snowflake.sql):

Snowflake-specific version with VARIANT type for metadata
TIMESTAMP_NTZ for Snowflake timestamp handling
Separate CREATE INDEX statements for Snowflake syntax

6. Comprehensive Tests (✅ Completed)

Test Suite (tests/test_events.py):

18+ test cases covering all event types, hooks, and EventBus functionality
Tests for event creation and metadata population
Tests for EventBus registration, emission, and hook execution
Tests for hook failure handling (failures don't stop other hooks)
Tests for built-in hooks (LoggingAlertHook, SQLEventHook)
Integration tests with in-memory SQLite database
Mock-based tests for external dependencies

7. Documentation and Examples (✅ Completed)

Comprehensive Documentation (EVENTS_AND_HOOKS.md):

Overview and core concepts
Detailed event type documentation with examples
Built-in hooks documentation
Configuration guide with examples
Custom hook creation guide
Usage examples for various scenarios
Best practices and troubleshooting
Integration with orchestrators (Dagster, Airflow)

Updated Configuration Examples:

examples/config.yml - Added hooks configuration section
Examples for logging, SQL, Snowflake, and custom hooks
Commented examples for easy copy-paste

Code Examples:

examples/example_hooks.py - 5 comprehensive examples demonstrating:
- Basic EventBus usage with logging
- Custom event collector hook
- Filtered alert hook (by severity)
- Multiple hooks working together
- Error handling (failures don't stop other hooks)

Updated Quickstart:

examples/quickstart.py - Added EventBus initialization
Demonstrates hook configuration loading
Shows EventBus integration with ProfileEngine and DriftDetector

Updated README (README.md):

Added event system to features list
New "Event & Alert Hooks" section with:
- Built-in hooks overview
- Configuration examples
- Event types
- Custom hooks guide
- Link to comprehensive documentation

Key Features

1. Orchestration-Agnostic

No external dependencies in core library
Works with or without orchestrators
Hooks handle external integrations

2. Failure-Resilient

Hook failures are caught and logged
One failing hook doesn't prevent others from executing
Profiling continues even if hooks fail

3. Highly Extensible

Simple AlertHook protocol for custom implementations
Dynamic custom hook loading from any module
Configuration-driven hook registration

4. Flexible Configuration

Master switch to enable/disable all hooks
Per-hook enable/disable flags
Environment-specific configurations
Environment variable support

5. Multiple Persistence Options

In-memory (logging)
SQL databases (Postgres, MySQL, SQLite)
Snowflake data warehouse
Custom destinations (webhooks, message queues, etc.)

Usage Examples

Basic Configuration

hooks:
  enabled: true
  hooks:
    - type: logging
      log_level: INFO

Production Configuration with Persistence

hooks:
  enabled: true
  hooks:
    # Log for immediate visibility
    - type: logging
      log_level: WARNING
    
    # Persist for historical analysis
    - type: snowflake
      table_name: prod.monitoring.baselinr_events
      connection:
        type: snowflake
        account: ${SNOWFLAKE_ACCOUNT}
        database: monitoring
        warehouse: compute_wh
        username: ${SNOWFLAKE_USER}
        password: ${SNOWFLAKE_PASSWORD}

Custom Slack Alert Hook

# my_hooks.py
import requests
from baselinr.events import BaseEvent, DataDriftDetected

class SlackAlertHook:
    def __init__(self, webhook_url: str, min_severity: str = "high"):
        self.webhook_url = webhook_url
        self.min_severity = min_severity
    
    def handle_event(self, event: BaseEvent) -> None:
        if not isinstance(event, DataDriftDetected):
            return
        
        if event.drift_severity != self.min_severity:
            return
        
        message = {
            "text": f"🚨 {event.drift_severity.upper()} drift in {event.table}.{event.column}"
        }
        requests.post(self.webhook_url, json=message)

hooks:
  enabled: true
  hooks:
    - type: custom
      module: my_hooks
      class_name: SlackAlertHook
      params:
        webhook_url: https://hooks.slack.com/services/YOUR/WEBHOOK
        min_severity: high

Architecture

Baselinr Core
    │
    ├─→ ProfileEngine
    │   ├─→ emit(ProfilingStarted)
    │   ├─→ emit(ProfilingCompleted)
    │   └─→ emit(ProfilingFailed)
    │
    └─→ DriftDetector
        ├─→ emit(DataDriftDetected)
        └─→ emit(SchemaChangeDetected)
                │
                ↓
            EventBus
                │
                ├─→ LoggingAlertHook → stdout
                ├─→ SQLEventHook → Postgres/SQLite
                ├─→ SnowflakeEventHook → Snowflake
                └─→ CustomHook → Your Integration

Testing

Run the comprehensive test suite:

# Run all event system tests
pytest tests/test_events.py -v

# Run with coverage
pytest tests/test_events.py --cov=baselinr.events --cov-report=html

Example test output:

tests/test_events.py::TestBaseEvent::test_base_event_creation PASSED
tests/test_events.py::TestDataDriftDetected::test_drift_event_creation PASSED
tests/test_events.py::TestEventBus::test_emit_event_to_multiple_hooks PASSED
tests/test_events.py::TestEventBus::test_hook_failure_does_not_stop_other_hooks PASSED
...
==================== 18 passed in 0.52s ====================

Example Usage

Run Profiling with Hooks

# With logging hook (from config.yml)
baselinr profile --config examples/config.yml

# Output:
# [ALERT] ProfilingStarted: {'table': 'customers', 'run_id': '...'}
# [ALERT] ProfilingCompleted: {'table': 'customers', 'row_count': 1000, ...}

Run Drift Detection with Hooks

baselinr drift --config examples/config.yml --dataset customers

# Events emitted:
# - DataDriftDetected (for each drifted metric)
# - SchemaChangeDetected (for schema changes)

Run Examples

# Run comprehensive examples
python examples/example_hooks.py

# Output demonstrates:
# - Basic event emission
# - Custom collectors
# - Filtered alerts
# - Multiple hooks
# - Error handling

Files Created/Modified

New Files

baselinr/events/__init__.py
baselinr/events/events.py
baselinr/events/hooks.py
baselinr/events/event_bus.py
baselinr/events/builtin_hooks.py
baselinr/storage/schema_snowflake.sql
tests/test_events.py
EVENTS_AND_HOOKS.md
examples/example_hooks.py
EVENTS_IMPLEMENTATION_SUMMARY.md (this file)

Modified Files

baselinr/config/schema.py - Added HookConfig, HooksConfig
baselinr/config/__init__.py - Added exports
baselinr/cli.py - Added EventBus initialization
baselinr/drift/detector.py - Added event emission
baselinr/profiling/core.py - Added event emission
baselinr/storage/schema.sql - Added baselinr_events table
examples/config.yml - Added hooks configuration
examples/quickstart.py - Added EventBus usage
README.md - Added event system documentation

Benefits

For Developers

✅ Real-time visibility into profiling operations
✅ Easy debugging with logging hooks
✅ Historical event tracking for analysis
✅ Extensible for custom integrations

For Data Teams

✅ Automatic drift alerts when thresholds are exceeded
✅ Schema change notifications
✅ Profiling lifecycle tracking
✅ Integration with existing alert systems

For Production

✅ Configurable alert destinations
✅ Failure-resilient (hooks don't break profiling)
✅ Environment-specific configurations
✅ Performance-conscious (async-ready)

Future Enhancements

Potential future improvements:

Event Filtering: Configure which events each hook receives
Async Hooks: Native async/await support for non-blocking operations
Event Batching: Batch multiple events for efficient persistence
Retry Logic: Automatic retry for failed hook executions
Rate Limiting: Prevent alert fatigue with configurable limits
Event Streaming: Kafka/Kinesis integration for event streams
Hook Metrics: Track hook performance and failure rates
Event Replay: Replay historical events for testing/debugging

Conclusion

The event and alert hook system is now fully implemented and integrated into Baselinr. It provides a powerful, flexible, and extensible way to react to profiling and drift detection events, enabling real-time alerts, historical tracking, and custom integrations.

The system is:

✅ Production-ready
✅ Well-tested (18+ test cases)
✅ Fully documented
✅ Backward compatible (hooks are optional)
✅ Easy to use and extend

For more information, see:

EVENTS_AND_HOOKS.md - Comprehensive documentation
examples/example_hooks.py - Code examples
tests/test_events.py - Test suite

Overview​

What Was Implemented​

1. Core Event System (✅ Completed)​

2. Integration with Core Components (✅ Completed)​

3. Configuration System (✅ Completed)​

4. CLI Integration (✅ Completed)​

5. Database Schema (✅ Completed)​

6. Comprehensive Tests (✅ Completed)​

7. Documentation and Examples (✅ Completed)​

Key Features​

1. Orchestration-Agnostic​

2. Failure-Resilient​

3. Highly Extensible​

4. Flexible Configuration​

5. Multiple Persistence Options​

Usage Examples​

Basic Configuration​

Production Configuration with Persistence​

Custom Slack Alert Hook​

Architecture​

Testing​

Example Usage​

Run Profiling with Hooks​

Run Drift Detection with Hooks​

Run Examples​

Files Created/Modified​

New Files​

Modified Files​

Benefits​

For Developers​

For Data Teams​

For Production​

Future Enhancements​

Conclusion​