Quick Start Tutorial

Get up and running with CallPyBack in minutes! This tutorial covers the essential patterns you'll use in 90% of applications.

📦 Installation

pip install callpyback

🎯 Your First CallPyBack Function

Let's start with the simplest possible example:

from callpyback import CallPyBack

@CallPyBack()
def hello_world():
    return "Hello, CallPyBack!"

result = hello_world()
print(result)  # "Hello, CallPyBack!"

What happened? CallPyBack decorated your function but didn't change its behavior. The real power comes when you add observers!

🔍 Adding Your First Observer

Let's add some basic logging:

from callpyback import CallPyBack, on_success

def log_success(result):
    print(f"✅ Function succeeded with result: {result.value}")

@CallPyBack(observers=[on_success(log_success)])
def calculate(x, y):
    return x + y

result = calculate(5, 3)
# Output: ✅ Function succeeded with result: 8
print(f"Result: {result}")  # Result: 8

Key concepts:

• Observers: Functions that monitor your decorated function
• on_success: Factory function that creates observers for successful executions
• result.value: The actual return value from your function

🛡️ Error Handling

Let's add error handling to make our functions more robust:

from callpyback import CallPyBack, on_success, on_failure

def log_success(result):
    print(f"✅ Success: {result.value}")

def log_error(result):
    print(f"❌ Error: {result.exception}")

@CallPyBack(
    observers=[on_success(log_success), on_failure(log_error)],
    exception_classes=(ValueError, TypeError),  # Catch these exceptions
    default_return=0  # Return this value on error
)
def safe_divide(a, b):
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

# Test successful execution
result = safe_divide(10, 2)
# Output: ✅ Success: 5.0
print(f"Result: {result}")  # Result: 5.0

# Test error handling
result = safe_divide(10, 0)
# Output: ❌ Error: Cannot divide by zero
print(f"Result: {result}")  # Result: 0 (default_return)

Key concepts:

• exception_classes: Tuple of exception types to catch
• default_return: Value to return when an exception is caught
• on_failure: Observer for failed executions

📊 Performance Monitoring

Monitor function performance with built-in observers:

import time
from callpyback import CallPyBack
from callpyback.observers.builtin import MetricsObserver, TimingObserver

# Create monitoring observers
metrics = MetricsObserver()
timing = TimingObserver(threshold=0.1)  # Alert if execution > 100ms

@CallPyBack(observers=[metrics, timing])
def slow_function(duration):
    time.sleep(duration)
    return f"Slept for {duration} seconds"

# Execute function multiple times
slow_function(0.05)  # Fast execution
slow_function(0.15)  # Slow execution (will trigger timing alert)
slow_function(0.02)  # Fast execution

# Get performance metrics
performance_data = metrics.get_metrics()
print(f"Total executions: {performance_data['total_executions']}")
print(f"Average time: {performance_data['average_execution_time']*1000:.1f}ms")

# Check slow executions
slow_executions = timing.get_slow_executions()
print(f"Slow executions detected: {len(slow_executions)}")

Key concepts:

• MetricsObserver: Collects execution statistics
• TimingObserver: Detects slow executions
• Built-in observers: Ready-to-use monitoring components

🔬 Variable Extraction

Capture local variables during function execution:

from callpyback import CallPyBack, on_completion

captured_variables = []

def capture_vars(local_variables):
    captured_variables.append(local_variables)
    print(f"📋 Captured variables: {local_variables}")

@CallPyBack(
    observers=[on_completion(capture_vars)],
    variable_names=["step1", "step2", "final_result"]
)
def data_processing(input_data):
    step1 = input_data.upper()
    step2 = step1.replace(" ", "_")
    final_result = f"processed_{step2}"
    return final_result

result = data_processing("hello world")
# Output: 📋 Captured variables: {'step1': 'HELLO WORLD', 'step2': 'HELLO_WORLD', 'final_result': 'processed_HELLO_WORLD'}
print(f"Result: {result}")

Key concepts:

• variable_names: List of local variable names to capture
• on_completion: Observer that always executes (success or failure)
• Variable extraction: Powerful debugging and auditing capability

🎯 Observer States

Control when observers execute by subscribing to specific states:

from callpyback import CallPyBack, 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"❌ Failed: {result.exception}")

def always_runs(context):
    print(f"🏁 Finished with state: {context.state.name}")

@CallPyBack(
    observers=[
        on_call(before_execution),      # Before execution
        on_success(after_success),      # Only on success
        on_failure(after_failure),      # Only on failure  
        on_completion(always_runs)      # Always executes
    ],
    exception_classes=(ValueError,),
    default_return="error_handled"
)
def stateful_function(should_fail=False):
    if should_fail:
        raise ValueError("Intentional error")
    return "success_result"

# Test successful execution
print("=== Testing Success ===")
stateful_function(False)
# Output:
# 🚀 Starting stateful_function
# ✅ Success: success_result
# 🏁 Finished with state: COMPLETED

print("\n=== Testing Failure ===")
stateful_function(True)
# Output:
# 🚀 Starting stateful_function
# ❌ Failed: Intentional error
# 🏁 Finished with state: COMPLETED

Key concepts:

• on_call: Executes before function starts
• State-based observers: Different observers for different execution phases
• context: Rich information about function execution

🏭 Production Example

Here's a more realistic example combining multiple features:

import time
import random
from callpyback import CallPyBack, on_call, on_success, on_failure
from callpyback.observers.builtin import MetricsObserver, LoggingObserver

# Setup monitoring
metrics = MetricsObserver(priority=100)
logger = LoggingObserver(priority=50)

# Track business metrics
business_metrics = {"orders_processed": 0, "total_revenue": 0}

def track_business_metrics(result):
    if isinstance(result.value, dict) and "revenue" in result.value:
        business_metrics["orders_processed"] += 1
        business_metrics["total_revenue"] += result.value["revenue"]

def log_order_start(context):
    print(f"📦 Processing order: {context.arguments}")

def handle_order_failure(result):
    print(f"🚨 Order processing failed: {result.exception}")
    # Could trigger alerts, create support tickets, etc.

@CallPyBack(
    observers=[
        metrics,                                # Built-in metrics
        logger,                                 # Built-in logging
        on_call(log_order_start),              # Business logging
        on_success(track_business_metrics),    # Business metrics
        on_failure(handle_order_failure)       # Error handling
    ],
    variable_names=["order_total", "processing_fee"],
    exception_classes=(ValueError, ConnectionError),
    default_return={"status": "failed", "order_id": None}
)
def process_order(customer_id, items, payment_method):
    # Simulate order processing
    order_total = sum(item["price"] for item in items)
    processing_fee = order_total * 0.03
    
    # Simulate potential failures
    if random.random() < 0.1:  # 10% failure rate
        raise ConnectionError("Payment gateway unavailable")
    
    if order_total <= 0:
        raise ValueError("Invalid order total")
    
    # Simulate processing time
    time.sleep(random.uniform(0.01, 0.05))
    
    return {
        "status": "completed",
        "order_id": f"ORD_{random.randint(1000, 9999)}",
        "revenue": order_total,
        "fee": processing_fee
    }

# Process some orders
orders = [
    (1001, [{"item": "laptop", "price": 999.99}], "credit_card"),
    (1002, [{"item": "mouse", "price": 29.99}, {"item": "keyboard", "price": 79.99}], "paypal"),
    (1003, [{"item": "monitor", "price": 299.99}], "credit_card"),
    (1004, [], "credit_card"),  # Invalid order - empty items
    (1005, [{"item": "tablet", "price": 499.99}], "credit_card")
]

print("=== Processing Orders ===")
for customer_id, items, payment_method in orders:
    result = process_order(customer_id, items, payment_method)
    print(f"Order result: {result}")
    print()

# Show final metrics
print("=== Final Metrics ===")
perf_metrics = metrics.get_metrics()
print(f"Total orders attempted: {perf_metrics['total_executions']}")
print(f"Success rate: {(perf_metrics['function_stats']['process_order']['successes'] / perf_metrics['function_stats']['process_order']['calls'] * 100):.1f}%")
print(f"Average processing time: {perf_metrics['average_execution_time']*1000:.1f}ms")
print(f"Orders completed: {business_metrics['orders_processed']}")
print(f"Total revenue: ${business_metrics['total_revenue']:.2f}")

🚀 Next Steps

Congratulations! You've learned the fundamentals of CallPyBack. Here's what to explore next:

📚 Learn More About:

• Observer System - Deep dive into observer patterns
• Custom Observers - Build domain-specific monitors
• Performance Monitoring - Production monitoring strategies
• Error Handling - Advanced exception management

🎯 Try These Patterns:

• API Monitoring: Monitor HTTP request/response cycles
• Database Auditing: Track query performance and access patterns
• ML Pipeline Monitoring: Monitor training and inference workflows
• Financial Transaction Auditing: Compliance and fraud detection

🔧 Advanced Features:

• Circuit Breaker Pattern - Fault tolerance
• Distributed Systems - Microservices integration
• Thread Safety - Concurrent execution patterns
• Security Auditing - Compliance and audit logging

💡 Pro Tips

1. Start Simple

Begin with basic observers (on_success, on_failure) and gradually add complexity.

2. Use Built-in Observers

The built-in observers (MetricsObserver, LoggingObserver, TimingObserver) cover most production needs.

3. Think About Priorities

Assign logical priorities to your observers:

• Security/Audit: 800-999
• Business Logic: 500-799
• Monitoring: 100-499
• Debug/Dev: 1-99

4. Variable Extraction is Powerful

Use variable extraction for debugging, auditing, and compliance. It's like having a debugger built into production code.

5. Error Handling Strategy

Always define exception_classes and default_return for production functions. This ensures graceful degradation.

🔗 Common Patterns

Web Application Monitoring

@CallPyBack(
    observers=[
        on_call(lambda ctx: log_request(ctx.arguments)),
        on_success(lambda result: log_response(result.value)),
        on_failure(lambda result: log_error(result.exception))
    ]
)
def api_endpoint(request_data):
    return process_request(request_data)

Database Operation Tracking

@CallPyBack(
    observers=[timing_observer, metrics_observer],
    variable_names=["query", "rows_affected"]
)
def database_operation(table, operation):
    query = f"SELECT * FROM {table}"
    rows_affected = execute_query(query)
    return rows_affected

Batch Processing Monitoring

@CallPyBack(
    observers=[progress_observer, metrics_observer],
    variable_names=["processed_count", "error_count"]
)
def batch_processor(items):
    processed_count = 0
    error_count = 0
    
    for item in items:
        try:
            process_item(item)
            processed_count += 1
        except Exception:
            error_count += 1
    
    return {"processed": processed_count, "errors": error_count}

---

Ready for more? Check out the Observer System documentation to understand the powerful patterns behind CallPyBack!