Observer System

The Observer System is the heart of CallPyBack, implementing a formal observer pattern that allows multiple components to monitor function execution without tight coupling.

🎯 Overview

The observer pattern in CallPyBack provides:

Decoupled Monitoring: Observers operate independently without affecting function execution
Thread-Safe Operations: Concurrent observer notification with error isolation
Priority-Based Execution: Higher priority observers execute first
State-Aware Filtering: Observers can subscribe to specific execution states
Error Isolation: Observer failures don't impact function execution or other observers

🏗️ Architecture


┌─────────────────────────────────────────────────────────────┐
│                    Observer Pattern                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────┐      ┌─────────────────────────────┐   │
│  │   CallPyBack    │      │     Observer Manager        │   │
│  │   Decorator     │────▶│                             │   │
│  │                 │      │ • Add/Remove Observers      │   │
│  │ • Function Exec │      │ • Priority Sorting          │   │
│  │ • State Changes │      │ • Thread-Safe Notification  │   │
│  │ • Error Handling│      │ • Error Isolation           │   │
│  └─────────────────┘      └─────────────────────────────┘   │
│           │                              │                  │
│           │ ExecutionContext             │ notify()         │
│           ▼                              ▼                  │
│  ┌─────────────────────────────────────────────────────┐    │
│  │                 Observers                           │    │
│  │                                                     │    │
│  │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐     │    │
│  │ │  Observer1  │ │  Observer2  │ │  Observer3  │     │    │
│  │ │ Priority:100│ │ Priority: 50│ │ Priority: 10│     │    │
│  │ │ States: ALL │ │ States:SUCC │ │ States:FAIL │     │    │
│  │ └─────────────┘ └─────────────┘ └─────────────┘     │    │
│  └─────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘

🔧 Observer Interface

All observers must implement the Observer protocol:

from callpyback.protocols import Observer
from callpyback.core.context import ExecutionContext
class MyObserver:
def init(self, priority: int = 0, name: str = "MyObserver"):
self.priority = priority
self.name = name
def update(self, context: ExecutionContext) -&gt; None:
    """Handle execution context update."""
    print(f"Observer {self.name} received update: {context.state}")

Required Attributes

Attribute	Type	Description
priority	int	Execution priority (higher = earlier)
name	str	Observer identifier for debugging

📊 Execution States

Observers can subscribe to specific execution states:

from callpyback.core.state_machine import ExecutionState
Available states
INITIALIZED     # Function decoration complete
PRE_EXECUTION   # Before function execution
EXECUTING       # During function execution
POST_SUCCESS    # After successful execution
POST_FAILURE    # After failed execution
COMPLETED       # Final state (always reached)
ERROR           # Unexpected error state

State Transition Flow

INITIALIZED
     │
     ▼
PRE_EXECUTION
     │
     ▼
EXECUTING ─────┬─── POST_SUCCESS ─── COMPLETED
               │
               └─── POST_FAILURE ──── COMPLETED
               │
               └─── ERROR

🏭 Observer Types

1. Built-in Observers

CallPyBack provides several production-ready observers:

MetricsObserver

from callpyback.observers.builtin import MetricsObserver
metrics = MetricsObserver(priority=100)
@CallPyBack(observers=[metrics])
def monitored_function():
return "result"
Get metrics
data = metrics.get_metrics()
print(f"Total executions: {data['total_executions']}")

LoggingObserver

from callpyback.observers.builtin import LoggingObserver
import logging
logger = LoggingObserver(
logger=logging.getLogger("myapp"),
log_level=logging.INFO,
priority=50
)
@CallPyBack(observers=[logger])
def logged_function():
return "result"

TimingObserver

from callpyback.observers.builtin import TimingObserver
timing = TimingObserver(
threshold=0.1,  # 100ms threshold
priority=75
)
@CallPyBack(observers=[timing])
def timed_function():
time.sleep(0.2)  # Will trigger slow execution alert
return "result"

2. Callback Observers

Simple function-based observers using the factory pattern:

from callpyback import on_call, on_success, on_failure, on_completion
def before_execution(context):
print(f"Starting {context.function_signature.name}")
def after_success(result):
print(f"Success: {result.value}")
def after_failure(result):
print(f"Error: {result.exception}")
def always_execute(context):
print(f"Completed with state: {context.state}")
@CallPyBack(observers=[
on_call(before_execution),
on_success(after_success),
on_failure(after_failure),
on_completion(always_execute)
])
def example_function():
return "hello world"

3. Custom Observers

Create domain-specific observers by extending BaseObserver:

from callpyback.observers.base import BaseObserver
from callpyback.core.state_machine import ExecutionState
class DatabaseAuditObserver(BaseObserver):
def init(self):
super().init(priority=90, name="DatabaseAudit")
self.queries = []
self.connection_pool = None
def update(self, context):
    if context.state == ExecutionState.COMPLETED:
        query_info = {
            'function': context.function_signature.name,
            'timestamp': context.timestamp,
            'success': context.is_successful,
            'execution_time': getattr(context.result, 'execution_time', 0)
        }
        
        # Log to database
        self._log_to_database(query_info)
        
        # Keep in-memory for analysis
        self.queries.append(query_info)

def _log_to_database(self, query_info):
    # Implementation for database logging
    pass

Usage
db_audit = DatabaseAuditObserver()
@CallPyBack(observers=[db_audit])
def database_operation():
return "query result"

⚙️ Observer Configuration

Priority System

Observers execute in priority order (highest first):

high_priority = MyObserver(priority=100)    # Executes first
medium_priority = MyObserver(priority=50)   # Executes second  
low_priority = MyObserver(priority=10)      # Executes last
@CallPyBack(observers=[low_priority, high_priority, medium_priority])
def prioritized_function():
return "result"

State Filtering

Subscribe observers to specific execution states:

# Manual state registration
decorator = CallPyBack()
Observer only for successful executions
decorator.add_observer(
success_observer,
states={ExecutionState.POST_SUCCESS}
)
Observer for both success and failure
decorator.add_observer(
completion_observer,
states={ExecutionState.POST_SUCCESS, ExecutionState.POST_FAILURE}
)
@decorator
def stateful_function():
return "result"

Error Isolation

Observer errors are automatically isolated:

def failing_observer(context):
    raise RuntimeError("Observer error")
def working_observer(context):
print("This observer still works!")
@CallPyBack(observers=[failing_observer, working_observer])
def resilient_function():
return "result"  # Function succeeds despite observer failure

🧵 Thread Safety

Concurrent Observer Manager

CallPyBack uses thread-safe observer management:

from concurrent.futures import ThreadPoolExecutor
import threading
metrics = MetricsObserver()
@CallPyBack(observers=[metrics])
def concurrent_function(worker_id):
time.sleep(0.01)  # Simulate work
return f"Worker {worker_id} completed"
Execute concurrently
with ThreadPoolExecutor(max_workers=10) as executor:
futures = [
executor.submit(concurrent_function, i)
for i in range(100)
]
results = [f.result() for f in futures]

All executions are safely tracked
print(f"Total executions: {metrics.get_metrics()['total_executions']}")

Error Isolating Observer Manager

Advanced error isolation with circuit breaker pattern:

from callpyback.management.observer_manager import ErrorIsolatingObserverManager
Create decorator with error isolation
decorator = CallPyBack(max_observer_failures=3)
Failing observer will be disabled after 3 failures
class FlakyObserver(BaseObserver):
def init(self):
super().init(priority=50, name="FlakyObserver")
self.call_count = 0
def update(self, context):
    self.call_count += 1
    if self.call_count % 2 == 0:  # Fail every other call
        raise RuntimeError("Flaky observer error")
    print(f"Observer call {self.call_count} succeeded")

flaky = FlakyObserver()
decorator.add_observer(flaky)
@decorator
def protected_function():
return "result"
Execute multiple times - observer gets disabled
for i in range(10):
protected_function()

📊 Monitoring and Debugging

Observer Metrics

Track observer performance and health:

# Get observer statistics
decorator = CallPyBack()
metrics = decorator.get_metrics()
print(f"Active observers: {metrics['observer_count']}")
Check observer manager state
manager = decorator._observer_manager
if hasattr(manager, '_disabled_observers'):
print(f"Disabled observers: {len(manager._disabled_observers)}")

Debug Observer

Create observers for debugging and development:

class DebugObserver(BaseObserver):
    def __init__(self):
        super().__init__(priority=999, name="Debug")  # Highest priority
def update(self, context):
    print(f"""
    🐛 DEBUG INFO:
    Function: {context.function_signature.name}
    State: {context.state.name}
    Arguments: {context.arguments}
    Success: {context.is_successful if context.result else 'N/A'}
    Timestamp: {context.timestamp}
    """)

Enable debug mode
debug = DebugObserver()
@CallPyBack(observers=[debug])
def debugged_function(x, y):
return x + y

🎯 Best Practices

1. Observer Design

Single Responsibility: Each observer should have one clear purpose
Error Resilience: Handle exceptions gracefully within observers
Minimal Overhead: Keep observer logic lightweight
Thread Safety: Use locks when modifying shared state

2. Priority Assignment

# Recommended priority ranges
CRITICAL_OBSERVERS = 900-999    # Security, audit logging
BUSINESS_OBSERVERS = 800-899    # Domain-specific logic  
MONITORING_OBSERVERS = 100-199  # Metrics, performance
DEBUG_OBSERVERS = 1-99          # Development, testing

3. State Selection

Use PRE_EXECUTION for setup and validation
Use POST_SUCCESS for success-specific logic
Use POST_FAILURE for error handling and cleanup
Use COMPLETED for always-executing cleanup

4. Memory Management

# Use weak references for large observer sets
from weakref import WeakSet
class LargeScaleObserver(BaseObserver):
def init(self):
super().init(priority=50, name="LargeScale")
self.tracked_functions = WeakSet()  # Won't prevent GC
def update(self, context):
    self.tracked_functions.add(context.function_signature.name)

🔗 Related Documentation

State Machine - Understanding execution flow
Built-in Observers - Ready-to-use observer implementations
Custom Observers - Building domain-specific monitors
Error Handling - Exception management strategies
Performance Monitoring - Production monitoring patterns

Next: Learn about the State Machine that drives observer notifications.

Observer System - adnanhd/observer-pattern GitHub Wiki

Observer System

🎯 Overview

🏗️ Architecture

🔧 Observer Interface

Required Attributes

📊 Execution States

Available states

State Transition Flow

🏭 Observer Types

1. Built-in Observers

MetricsObserver

Get metrics

LoggingObserver

TimingObserver

2. Callback Observers

3. Custom Observers

Usage

⚙️ Observer Configuration

Priority System

State Filtering

Observer only for successful executions

Observer for both success and failure

Error Isolation

🧵 Thread Safety

Concurrent Observer Manager

Execute concurrently

All executions are safely tracked

Error Isolating Observer Manager

Create decorator with error isolation

Failing observer will be disabled after 3 failures

Execute multiple times - observer gets disabled

📊 Monitoring and Debugging

Observer Metrics

Check observer manager state