Library

Beaker

Overview

Beaker is a session management and caching library for web applications written in Python.

Details

Beaker is a comprehensive caching and session library for web applications and standalone Python scripts and applications. It operates as WSGI middleware and provides simple session management by abstracting session implementation details. All cookies are protected with HMAC signatures to prevent tampering, and sessions are only loaded from the file system when session objects are actually accessed, preventing performance degradation on pages that don't use sessions. It supports multiple backend storage options (file, dbm, memory, memcached, database, google, mongodb, redis) with flexible configuration. For sessions, an additional cookie type is available that can store all session data in the cookie itself (with a 4096-byte limit). While widely known for use with the Pylons framework, it can be used with any WSGI-compatible framework.

Pros and Cons

Pros

• WSGI Integration: Easy integration as standard WSGI middleware
• Rich Backends: Supports 8 different storage backends
• Security: Cookie tampering prevention through HMAC signatures
• Performance: Efficient session management through lazy loading
• Flexibility: Provides both caching and session functionality
• Easy Configuration: Intuitive configuration options
• Lightweight: Minimal dependencies

Cons

• Old Design: Relatively old library with limited modern features
• Maintenance: Active development has decreased
• Security Warnings: Potential security risks from pickle module usage
• Python2 Era: Some features are outdated due to Python3 migration
• Alternative Libraries: More modern caching and session libraries exist

Key Links

• Beaker Documentation
• Beaker PyPI
• Beaker GitHub
• Pylons Documentation
• WSGI Specification

Code Examples

Installation

pip install Beaker

Basic WSGI Middleware Setup

from beaker.middleware import SessionMiddleware

# WSGI application definition
def simple_app(environ, start_response):
    session = environ['beaker.session']
    
    # Get data from session
    count = session.get('count', 0)
    count += 1
    
    # Save data to session
    session['count'] = count
    session.save()
    
    response_body = f'This is your {count} visit'
    status = '200 OK'
    headers = [('Content-Type', 'text/plain; charset=utf-8')]
    
    start_response(status, headers)
    return [response_body.encode('utf-8')]

# Session middleware configuration
session_opts = {
    'session.type': 'file',
    'session.cookie_expires': 300,
    'session.data_dir': './data',
    'session.auto': True
}

app = SessionMiddleware(simple_app, session_opts)

Flask Integration

from flask import Flask, session, request
from beaker.middleware import SessionMiddleware

app = Flask(__name__)

# Beaker session configuration
session_opts = {
    'session.type': 'ext:redis',
    'session.url': 'redis://localhost:6379',
    'session.cookie_expires': 3600,
    'session.auto': True,
    'session.encrypt_key': 'your-secret-key-here',
    'session.validate_key': 'your-validation-key-here'
}

@app.route('/')
def index():
    beaker_session = request.environ['beaker.session']
    visits = beaker_session.get('visits', 0)
    visits += 1
    beaker_session['visits'] = visits
    beaker_session.save()
    
    return f'Visit count: {visits}'

@app.route('/user/<username>')
def user_profile(username):
    beaker_session = request.environ['beaker.session']
    beaker_session['current_user'] = username
    beaker_session.save()
    
    return f'Welcome, {username}!'

@app.route('/logout')
def logout():
    beaker_session = request.environ['beaker.session']
    beaker_session.delete()
    return 'Logged out'

# Wrap application with Beaker middleware
app.wsgi_app = SessionMiddleware(app.wsgi_app, session_opts)

if __name__ == '__main__':
    app.run(debug=True)

Using Cache Functionality

from beaker.cache import CacheManager
from beaker.util import parse_cache_config_options

# Cache configuration
cache_opts = {
    'cache.type': 'file',
    'cache.data_dir': '/tmp/cache/data',
    'cache.lock_dir': '/tmp/cache/lock'
}

cache = CacheManager(**parse_cache_config_options(cache_opts))

# Basic cache operations
def expensive_function(param):
    """Simulate heavy computation"""
    import time
    time.sleep(2)  # 2-second processing
    return f"Calculation result: {param * 2}"

# Using cache decorator
@cache.cache('expensive_func', expire=300)  # Cache for 5 minutes
def cached_expensive_function(param):
    return expensive_function(param)

# Manual cache operations
def manual_cache_example():
    # Create cache namespace
    mycache = cache.get_cache('manual_cache', expire=600)
    
    # Check if key exists
    key = 'user_data_123'
    if key in mycache:
        return mycache[key]
    
    # Calculate data and save to cache
    data = fetch_user_data_from_db(123)
    mycache[key] = data
    return data

def fetch_user_data_from_db(user_id):
    """Fetch user data from database (simulated)"""
    import time
    time.sleep(1)
    return {'id': user_id, 'name': f'User {user_id}', 'email': f'user{user_id}@example.com'}

Advanced Session Configuration

import json
from beaker.middleware import SessionMiddleware

# Custom serializer configuration
def json_serializer(obj):
    return json.dumps(obj)

def json_deserializer(data):
    return json.loads(data)

# Secure session configuration
secure_session_opts = {
    'session.type': 'ext:redis',
    'session.url': 'redis://localhost:6379/1',
    'session.cookie_expires': True,  # Delete on browser close
    'session.cookie_domain': '.example.com',
    'session.cookie_path': '/',
    'session.cookie_secure': True,  # HTTPS required
    'session.cookie_httponly': True,  # No JavaScript access
    'session.encrypt_key': 'your-very-long-encryption-key',
    'session.validate_key': 'your-validation-key',
    'session.auto': True,
    'session.timeout': 3600,  # 1-hour timeout
    'session.data_serializer': 'json',  # Use JSON instead of pickle
}

class SecureSessionApp:
    def __init__(self):
        self.session_opts = secure_session_opts
        
    def __call__(self, environ, start_response):
        session = environ['beaker.session']
        
        # Session validation
        if not self.validate_session(session):
            session.delete()
            return self.unauthorized_response(start_response)
        
        # Normal processing
        return self.handle_request(environ, start_response, session)
    
    def validate_session(self, session):
        """Validate session validity"""
        if 'user_id' not in session:
            return False
        
        # Check last activity time
        last_activity = session.get('last_activity')
        if last_activity:
            import time
            if time.time() - last_activity > 3600:  # 1 hour
                return False
        
        # Update activity time
        session['last_activity'] = time.time()
        session.save()
        return True
    
    def handle_request(self, environ, start_response, session):
        user_id = session.get('user_id')
        response_body = f'Logged in as user {user_id}'
        
        status = '200 OK'
        headers = [('Content-Type', 'text/plain; charset=utf-8')]
        start_response(status, headers)
        return [response_body.encode('utf-8')]
    
    def unauthorized_response(self, start_response):
        status = '401 Unauthorized'
        headers = [('Content-Type', 'text/plain')]
        start_response(status, headers)
        return [b'Unauthorized']

# Create application
app = SecureSessionApp()
app = SessionMiddleware(app, secure_session_opts)

Multi-Storage Configuration

from beaker.cache import CacheManager
from beaker.util import parse_cache_config_options

# Multiple cache storage configuration
cache_config = {
    # Memory cache (fast, volatile)
    'cache.memory.type': 'memory',
    
    # File cache (persistent)
    'cache.file.type': 'file',
    'cache.file.data_dir': '/tmp/cache/data',
    'cache.file.lock_dir': '/tmp/cache/lock',
    
    # Redis cache (distributed)
    'cache.redis.type': 'ext:redis',
    'cache.redis.url': 'redis://localhost:6379/2',
    
    # Database cache
    'cache.database.type': 'ext:database',
    'cache.database.url': 'postgresql://user:pass@localhost/cache_db',
    'cache.database.table_name': 'beaker_cache'
}

cache_manager = CacheManager(**parse_cache_config_options(cache_config))

class MultiTierCache:
    def __init__(self, cache_manager):
        self.memory_cache = cache_manager.get_cache('memory_cache', type='memory', expire=300)
        self.file_cache = cache_manager.get_cache('file_cache', type='file', expire=3600)
        self.redis_cache = cache_manager.get_cache('redis_cache', type='ext:redis', expire=7200)
    
    def get(self, key):
        """Check cache hierarchically"""
        # L1: Memory cache
        if key in self.memory_cache:
            return self.memory_cache[key]
        
        # L2: File cache
        if key in self.file_cache:
            value = self.file_cache[key]
            self.memory_cache[key] = value  # Save to memory cache too
            return value
        
        # L3: Redis cache
        if key in self.redis_cache:
            value = self.redis_cache[key]
            self.file_cache[key] = value    # Save to file cache too
            self.memory_cache[key] = value  # Save to memory cache too
            return value
        
        return None
    
    def set(self, key, value):
        """Save data to all tiers"""
        self.memory_cache[key] = value
        self.file_cache[key] = value
        self.redis_cache[key] = value
    
    def delete(self, key):
        """Delete data from all tiers"""
        self.memory_cache.pop(key, None)
        self.file_cache.pop(key, None)
        self.redis_cache.pop(key, None)

# Usage example
multi_cache = MultiTierCache(cache_manager)

def cached_api_call(api_endpoint):
    """Optimize API calls with multi-tier cache"""
    cache_key = f"api:{api_endpoint}"
    
    # Try to get from cache
    cached_result = multi_cache.get(cache_key)
    if cached_result:
        return cached_result
    
    # Actually call API
    result = make_api_request(api_endpoint)
    
    # Cache to all tiers
    multi_cache.set(cache_key, result)
    
    return result

def make_api_request(endpoint):
    """Actual API call (simulated)"""
    import time
    time.sleep(0.5)  # Simulate network latency
    return {'data': f'Response from {endpoint}', 'timestamp': time.time()}