Wayne's Framework

A robust, modular Python framework for building automation and research applications. Extracted from high-performance automation tools, this framework provides the "engine" without domain-specific logic—designed as a starting point for new Python projects.

---

Overview

Wayne's Framework is a production-ready foundation for building applications that require:

• API integrations with intelligent rate limiting and retry logic
• Browser automation using undetected ChromeDriver
• Database operations with SQLAlchemy connection pooling
• Resilient error handling with circuit breakers and graceful degradation

The framework follows a session-orchestrated architecture where SessionManager coordinates specialized managers (Database, Browser, API) through a clean, maintainable interface.

---

User Instructions

What This Project Is

Wayne's Framework is a reusable Python foundation designed to accelerate development of automation projects. It handles the complex infrastructure concerns so you can focus on business logic:

• Menu-driven CLI for interactive action execution
• Action Registry pattern for clean separation of concerns
• Built-in testing infrastructure with formatted output
• Production-grade error handling with retries and fallbacks

Installation

1. Clone the repository:

   git clone https://github.com/yourusername/waynes_framework.git
   cd waynes_framework

2. Create a virtual environment (recommended):

   python -m venv venv
   # Windows
   venv\Scripts\activate
   # macOS/Linux
   source venv/bin/activate

3. Install dependencies:

   pip install -r requirements.txt

4. Configure environment (optional):

   cp .env.example .env
   # Edit .env with your settings

Quick Start

Run the framework:

python main.py

You'll see an interactive menu with example actions. Select an option by entering its number.

Creating Your First Action

1. Create a new file in actions/:

   # actions/my_actions.py
   from core.action_registry import (
       ActionMetadata, ActionCategory, ActionRequirement, register_action
   )
   
   def my_custom_action(session_manager):
       """My custom action logic."""
       print("Hello from my action!")
       # Use session_manager.db_manager for database operations
       # Use session_manager.api_manager for API calls
       # Use session_manager.browser_manager for browser automation
       return True
   
   register_action(ActionMetadata(
       id="10",
       name="My Custom Action",
       description="Does something amazing",
       category=ActionCategory.WORKFLOW,
       browser_requirement=ActionRequirement.NONE,
       function=my_custom_action,
   ))

2. Import in actions/__init__.py:

   from actions import my_actions  # noqa: F401

3. Run and test:

   python main.py

Running Tests

python run_tests.py

---

Developer Instructions

Architecture Overview

┌─────────────────────────────────────────────────────────────┐
│                        main.py                               │
│                   (CLI Entry Point)                          │
└─────────────────────────┬───────────────────────────────────┘
                          │
┌─────────────────────────▼───────────────────────────────────┐
│                   SessionManager                             │
│              (Central Orchestrator)                          │
│  ┌──────────────┬──────────────┬──────────────┬───────────┐ │
│  │DatabaseManager│BrowserManager│  APIManager  │RateLimiter│ │
│  └──────────────┴──────────────┴──────────────┴───────────┘ │
└─────────────────────────────────────────────────────────────┘

Project Structure

waynes_framework/
├── main.py                  # CLI entry point and menu system
├── run_tests.py             # Test runner script
├── requirements.txt         # Python dependencies
├── .env.example             # Environment configuration template
├── .gitignore               # Git ignore patterns
│
├── core/                    # Core framework components
│   ├── __init__.py          # Package exports
│   ├── session_manager.py   # Central orchestrator
│   ├── database_manager.py  # SQLAlchemy operations
│   ├── browser_manager.py   # WebDriver lifecycle
│   ├── api_manager.py       # HTTP request handling
│   ├── action_registry.py   # Action metadata system
│   ├── rate_limiter.py      # Adaptive rate limiting
│   └── error_handling.py    # Error recovery infrastructure
│
├── config/                  # Configuration management
│   ├── __init__.py
│   └── config_schema.py     # Type-safe config schemas
│
├── actions/                 # User-defined actions
│   ├── __init__.py
│   └── example_actions.py   # Example action implementations
│
├── browser/                 # Browser automation
│   ├── __init__.py
│   └── chromedriver.py      # ChromeDriver initialization
│
├── testing/                 # Testing infrastructure
│   ├── __init__.py
│   └── test_framework.py    # Test suite utilities
│
├── ai/                      # AI integrations (placeholder)
├── caching/                 # Caching utilities (placeholder)
├── models/                  # Data models (placeholder)
└── observability/           # Monitoring tools (placeholder)

Core Components

SessionManager (core/session_manager.py)

The central orchestrator that manages lifecycle state and coordinates all managers:

from core import SessionManager

session = SessionManager(db_path="data/myapp.db")

# Check if session is ready for operations
if session.ensure_session_ready("my_action"):
    # Session is ready - database initialized, browser available if needed
    pass

# Always close when done
session.close()

Lifecycle States:

• UNINITIALIZED - Initial state before any setup
• RECOVERING - Transitioning to ready state
• READY - All components initialized and healthy
• DEGRADED - Some components failed but partial operation possible

DatabaseManager (core/database_manager.py)

SQLAlchemy-based database manager with connection pooling:

from core import DatabaseManager

db = DatabaseManager(db_path="data/app.db")
db.ensure_ready()

# Batch operations for performance
with db.batch_operation_context(batch_size=1000) as (session, add_to_batch, commit_batch):
    for item in items:
        session.add(item)
        add_to_batch()  # Auto-commits every 1000 items

# Performance monitoring
stats = db.get_performance_stats()
print(f"Slow queries: {stats['slow_queries']}")

APIManager (core/api_manager.py)

Unified HTTP request interface with retries:

from core import APIManager, RequestConfig, RetryPolicy

api = APIManager()

result = api.request(RequestConfig(
    url="https://api.example.com/data",
    method="GET",
    headers={"Authorization": "Bearer token"},
    retry_policy=RetryPolicy.RESILIENT,
    max_retries=5,
))

if result.success:
    print(result.data)
else:
    print(f"Error: {result.error}")

AdaptiveRateLimiter (core/rate_limiter.py)

Token-bucket rate limiter that adapts to API pressure:

from core import AdaptiveRateLimiter

limiter = AdaptiveRateLimiter(
    initial_fill_rate=2.0,  # 2 requests/second
    min_fill_rate=0.1,      # Floor when throttled
    max_fill_rate=5.0,      # Ceiling when healthy
)

# Before each API call
limiter.wait("my_endpoint")

# After successful call
limiter.on_success("my_endpoint")

# After 429 error
limiter.on_429_error("my_endpoint", retry_after=60)

Error Handling (core/error_handling.py)

Comprehensive error recovery with decorators:

from core import (
    with_enhanced_recovery,
    graceful_degradation,
    timeout_protection,
    CircuitBreaker,
)

# Automatic retries with exponential backoff
@with_enhanced_recovery(max_attempts=3, base_delay=1.0)
def fetch_data():
    return api.get("/data")

# Fallback on failure
@graceful_degradation(fallback_value=[])
def get_items():
    return db.query(Item).all()

# Timeout protection
@timeout_protection(timeout=30)
def long_running_task():
    # Will raise TimeoutError if exceeds 30 seconds
    pass

# Circuit breaker pattern
breaker = CircuitBreaker("external_service")
result = breaker.call(external_api.fetch, params)

Environment Variables

Copy .env.example to .env and configure:

Variable	Default	Description
APP_NAME	WaynesFramework	Application identifier
DEBUG	false	Enable verbose logging
LOG_LEVEL	INFO	Logging level
DATABASE_PATH	data/app.db	SQLite database path
BROWSER_HEADLESS	false	Run browser headlessly
RATE_LIMIT_INITIAL	2.0	Initial requests/second
API_MAX_RETRIES	3	Maximum retry attempts

Adding New Features

1. New Manager: Add to core/, update core/__init__.py, inject via SessionManager
2. New Action Category: Add to ActionCategory enum in core/action_registry.py
3. New Error Type: Subclass FrameworkError, RetryableError, or FatalError
4. New Model: Add to models/ package with SQLAlchemy model definition

Testing Conventions

Use the built-in TestSuite for consistent test output:

from testing import TestSuite, suppress_logging

def run_my_tests():
    suite = TestSuite("My Feature Tests", "my_module.py")
    suite.start_suite()

    def test_something():
        result = my_function()
        assert result == expected

    suite.run_test(
        "Test Something",
        test_something,
        "Verify my_function returns expected value",
    )

    return suite.finish_suite()

Gotchas and Tips

1. Browser Manager: Browser isn't started by default. Set browser_needed=True or use ActionRequirement.DRIVER_ONLY / FULL_SESSION.

2. Rate Limiter Duplicate: The SessionManager initializes rate_limiter once. Don't re-assign it.

3. Database Path: Relative paths are relative to working directory. Use absolute paths for production.

4. Import Order: Import actions after the registry is available (main.py handles this).

5. Windows Compatibility: The timeout_protection decorator uses threading on Windows (no signal.SIGALRM).

---

Future Development Ideas

• Environment Loader: Add python-dotenv integration for .env file loading
• CLI Arguments: Add argparse for command-line action selection
• Plugin System: Dynamic action loading from external packages
• Web Interface: Optional Flask/FastAPI dashboard
• Async Support: Add async versions of managers for concurrent operations
• Prometheus Metrics: Add observability metrics export
• Redis Caching: Implement distributed caching in caching/ package
• LLM Integration: Add OpenAI/Anthropic clients in ai/ package
• Docker Support: Add Dockerfile and docker-compose.yml
• CI/CD Pipeline: Add GitHub Actions workflow

---

Appendix A: Chronology of Changes

Date	Version	Changes
2026-01-07	1.0.0	Initial framework release
		- Core components: SessionManager, DatabaseManager, BrowserManager, APIManager
		- Action registry with menu system
		- Adaptive rate limiter
		- Comprehensive error handling with circuit breakers
		- Testing infrastructure with 64 comprehensive tests
		- Example actions demonstrating framework usage
		- Package structure with init.py files
		- Configuration schema and .env.example

---

Appendix B: Technical Specifications

Python Version

• Minimum: Python 3.10+
• Tested: Python 3.11, 3.12

Dependencies

Package	Version	Purpose
sqlalchemy	≥2.0.0	Database ORM and connection pooling
requests	≥2.28.0	HTTP client for API operations
undetected-chromedriver	≥3.5.0	Stealth browser automation (optional)
selenium	≥4.10.0	WebDriver interface (optional)

Database Support

• Default: SQLite (file-based, no server required)
• Compatible: Any SQLAlchemy-supported database (PostgreSQL, MySQL, etc.)

Browser Support

• Primary: Google Chrome via undetected-chromedriver
• Mode: Headed (default) or headless

Rate Limiting Algorithm

• Type: Adaptive token bucket
• Adaptation: Rate decreases on 429 errors, increases on success streaks
• Per-endpoint: Separate rate tracking per endpoint label

Error Recovery Strategy

• Retryable Errors: Exponential backoff with jitter
• Fatal Errors: Immediate propagation (no retry)
• Circuit Breaker: Opens after 5 failures, half-opens after 60 seconds

Performance Characteristics

• Connection Pool: 10 base, 20 max overflow
• Query Timeout: Configurable (default 30s)
• Slow Query Threshold: 1.0 second

---

License

MIT License - See LICENSE file for details.