Gobstopper Application Core

The Gobstopper class is the central application object that orchestrates all framework functionality. It provides a Flask-like API while leveraging RSGI protocol for high performance and supporting optional Rust components for extreme speed.

Overview

Gobstopper is a hybrid Python/Rust web framework that automatically detects and uses performance-optimized Rust components when available, gracefully falling back to Python implementations. It’s built specifically for Granian’s RSGI interface and provides enterprise-grade features out of the box.

Gobstopper Class

Constructor

from gobstopper import Gobstopper

app = Gobstopper(name=__name__)

Parameters:

• name (str, optional): Application identifier, typically __name__ of the calling module. Used for logging and debugging.

Key Attributes:

• name: Application identifier

• logger: Structured logger with request context and tracing support

• routes: List of registered HTTP and WebSocket route handlers

• error_handlers: HTTP status code to error handler mapping

• middleware: Middleware functions with priority ordering

• template_engine: Template engine instance (None until initialized)

• task_queue: Background task queue with DuckDB persistence

• rust_router_available: Whether Rust routing components are available

HTTP Routing

Route Registration

Gobstopper provides decorators for all standard HTTP methods:

@app.get("/")
async def index(request):
    return {"message": "Hello World"}

@app.post("/users")
async def create_user(request):
    data = await request.json()
    return {"user": data}

@app.put("/users/<user_id>")
async def update_user(request, user_id):
    data = await request.json()
    return {"user_id": user_id, "updated": True}

@app.delete("/users/<user_id>")
async def delete_user(request, user_id):
    return {"deleted": user_id}

Supported HTTP Methods

Method	Decorator	Use Case
GET	@app.get(path)	Retrieve resources
POST	@app.post(path)	Create new resources
PUT	@app.put(path)	Update/replace resources
DELETE	@app.delete(path)	Remove resources
PATCH	@app.patch(path)	Partial resource updates
OPTIONS	@app.options(path)	CORS preflight, capability discovery

Trailing Slash Policy

Control how the router handles trailing slashes on routes. Configure via app.slash_policy:

• “add_slash”: Requests missing a trailing slash are redirected with 308 Permanent Redirect to the slash-suffixed URL.

• “remove_slash”: Requests with a trailing slash are redirected with 308 to the non-slashed URL.

• None (default): No automatic redirect; both forms can be registered explicitly.

Example:

app.slash_policy = "add_slash"  # or "remove_slash"

Note: Redirects use method-preserving 308 semantics.

Startup Diagnostics

On startup, Gobstopper performs diagnostics to help catch configuration issues early:

• Route conflict detection: detects duplicate static routes and dynamic/static shadowing and reports clear diagnostics.

• Blueprint hook signatures: validates before_request(request) and after_request(request, response) signatures with helpful error messages.

These diagnostics are logged and will raise clear errors during startup if misconfigured.

Path Parameters

Gobstopper supports dynamic path parameters with Flask-like syntax:

# Single parameter
@app.get("/users/<user_id>")
async def get_user(request, user_id):
    return {"user_id": user_id}

# Multiple parameters
@app.get("/users/<user_id>/posts/<post_id>")
async def get_user_post(request, user_id, post_id):
    return {"user_id": user_id, "post_id": post_id}

# Nested parameters
@app.get("/api/v<version>/users/<user_id>")
async def get_user_versioned(request, version, user_id):
    return {"version": version, "user_id": user_id}

Parameter Conversion: Path parameters are automatically extracted as strings. For type conversion:

@app.get("/users/<user_id>")
async def get_user(request, user_id):
    user_id = int(user_id)  # Manual conversion
    user = await get_user_by_id(user_id)
    return {"user": user}

Generic Route Registration

For complex routing needs, use the generic route() decorator:

# Multiple methods on same endpoint
@app.route("/api/resource", methods=['GET', 'POST', 'PUT'])
async def handle_resource(request):
    if request.method == 'GET':
        return await get_resource()
    elif request.method == 'POST':
        return await create_resource(await request.json())
    elif request.method == 'PUT':
        return await update_resource(await request.json())

Reverse Routing (url_for)

Gobstopper provides Flask/Quart-style reverse routing to build URLs from route names:

from gobstopper.http import redirect

# Named routes (explicit name)
@app.get('/users/<int:id>', name='user_detail')
async def get_user(request, id):
    return {"user": user_data}

# Automatic naming (uses function name)
@app.get('/dashboard')
async def dashboard(request):
    return {"dashboard": True}

# Build URLs
url = app.url_for('user_detail', id=123)  # Returns: '/users/123'
url = app.url_for('dashboard')            # Returns: '/dashboard'

# With redirects
@app.post('/users')
async def create_user(request):
    new_id = save_user()
    # Post-Redirect-Get pattern (status 303)
    return redirect(app.url_for('user_detail', id=new_id), status=303)

Blueprint Routes:

Blueprint routes are automatically qualified with the blueprint name:

from gobstopper.core.blueprint import Blueprint

admin = Blueprint('admin', __name__)

@admin.get('/login')
async def login(request):
    return render_template('admin/login.html')

@admin.get('/dashboard')
async def dashboard(request):
    return render_template('admin/dashboard.html')

app.register_blueprint(admin, url_prefix='/admin')

# Use blueprint-qualified names
app.url_for('admin.login')      # Returns: '/admin/login'
app.url_for('admin.dashboard')  # Returns: '/admin/dashboard'

# In route handlers
@app.get('/')
async def index(request):
    return redirect(app.url_for('admin.login'))

Redirect Status Codes:

# 302 - Temporary redirect (default)
return redirect('/new-page')

# 301 - Permanent redirect (cached by browsers)
return redirect('/new-location', status=301)

# 303 - See Other (POST → GET, recommended for form submissions)
return redirect(app.url_for('success'), status=303)

# 307 - Temporary redirect (preserves HTTP method)
return redirect('/new-endpoint', status=307)

# 308 - Permanent redirect (preserves HTTP method)
return redirect('/new-endpoint', status=308)

WebSocket Support

Gobstopper provides first-class WebSocket support through the RSGI protocol:

@app.websocket("/ws/echo")
async def websocket_echo(websocket):
    await websocket.accept()
    
    while True:
        message = await websocket.receive_text()
        await websocket.send_text(f"Echo: {message}")

@app.websocket("/ws/chat/<room_id>")
async def chat_room(websocket, room_id):
    await websocket.accept()
    
    # Join room logic
    await join_room(websocket, room_id)
    
    try:
        while True:
            message = await websocket.receive_text()
            await broadcast_to_room(room_id, message)
    finally:
        await leave_room(websocket, room_id)

Template Engine Integration

Initialization

Gobstopper supports both Jinja2 and high-performance Rust template engines:

# Auto-detection (uses Rust if available)
app.init_templates("templates")

# Force specific engine
app.init_templates("templates", use_rust=True)   # Rust only
app.init_templates("templates", use_rust=False)  # Jinja2 only

# With custom options
app.init_templates(
    "templates",
    auto_reload=False,      # Disable file watching
    cache_size=1000,        # Larger cache
    enable_streaming=True   # Rust streaming (if available)
)

Template Rendering

@app.get("/")
async def index(request):
    return await app.render_template("index.html", 
                                   title="Home Page",
                                   user=request.user)

@app.get("/dashboard")  
async def dashboard(request):
    users = await get_users()
    stats = await get_statistics()
    
    return await app.render_template("dashboard.html",
                                   users=users,
                                   stats=stats,
                                   page_title="Dashboard")

# Streaming for large datasets (Rust only)
@app.get("/report")
async def large_report(request):
    big_data = await get_large_dataset()
    return await app.render_template("report.html",
                                   stream=True,
                                   data=big_data)

Template Context Processors

Add global template variables:

@app.context_processor
def inject_globals():
    return {
        'app_name': 'MyApp',
        'current_year': datetime.now().year,
        'debug': app.config.get('DEBUG', False)
    }

# Async context processor
@app.context_processor
async def inject_user_data():
    return {
        'online_users': await count_online_users(),
        'system_status': await get_system_status()
    }

Template Filters and Globals

# Custom template filter
@app.template_filter('currency')
def currency_filter(value):
    return f"${value:.2f}"

# Custom template global
@app.template_global('get_config')
def get_config_value(key):
    return app.config.get(key)

# Usage in templates:
# {{ price | currency }}
# {{ get_config('API_URL') }}

Background Task System

Gobstopper includes a powerful background task queue with DuckDB persistence:

Task Registration

@app.task("send_email", category="notifications")
async def send_email(to: str, subject: str, body: str):
    # Send email logic
    await mail_service.send(to, subject, body)
    return {"status": "sent", "to": to}

@app.task("process_image", category="media")
async def process_image(image_path: str):
    # Image processing logic
    processed_path = await image_processor.process(image_path)
    return {"original": image_path, "processed": processed_path}

@app.task("generate_report", category="reports")
async def generate_report(user_id: int, report_type: str):
    # Long-running report generation
    report_data = await complex_report_generation(user_id, report_type)
    return {"report_id": report_data["id"], "status": "completed"}

Queueing Tasks

from gobstopper.tasks.queue import TaskPriority

@app.post("/send-email")
async def queue_email(request):
    data = await request.json()
    
    task_id = await app.add_background_task(
        "send_email",
        category="notifications",
        priority=TaskPriority.HIGH,
        max_retries=3,
        to=data["email"],
        subject=data["subject"],
        body=data["message"]
    )
    
    return {"task_id": task_id}

@app.get("/tasks/<task_id>")
async def get_task_status(request, task_id):
    status = await app.task_queue.get_task_status(task_id)
    return {"task_id": task_id, "status": status}

Starting Task Workers

@app.on_startup
async def start_background_workers():
    # Start workers for different categories
    await app.start_task_workers("notifications", worker_count=3)
    await app.start_task_workers("media", worker_count=2)
    await app.start_task_workers("reports", worker_count=1)

Middleware System

Gobstopper supports prioritized middleware for request/response processing:

Adding Middleware

from gobstopper.middleware.cors import CORSMiddleware
from gobstopper.middleware.security import SecurityMiddleware

# Add middleware with priority (higher numbers execute first)
app.add_middleware(SecurityMiddleware, priority=100)
app.add_middleware(CORSMiddleware, priority=90)

# Custom middleware
async def logging_middleware(request, next_handler):
    start_time = time.time()
    
    response = await next_handler(request)
    
    duration = time.time() - start_time
    app.logger.info(f"Request {request.path} took {duration:.3f}s")
    
    return response

app.add_middleware(logging_middleware, priority=50)

Built-in Middleware

Available middleware components:

• SecurityMiddleware: CSRF protection, security headers, session management

• CORSMiddleware: Cross-origin resource sharing configuration

• StaticMiddleware: Static file serving (Python implementation)

• RustStaticMiddleware: High-performance static file serving (Rust implementation)

Session Management

Gobstopper’s SecurityMiddleware provides a flexible session management system with pluggable storage backends.

Choosing a Backend

You can configure the session backend by passing a storage instance to the SecurityMiddleware.

• FileSessionStorage (default): Simple file-based storage. Not recommended for production, especially in ephemeral or multi-instance environments.

• MemorySessionStorage: In-memory storage, ideal for tests and development. Data is lost on application restart.

• AsyncRedisSessionStorage: Recommended for production. Requires a running Redis server and the redis extra (pip install gobstopper[redis]).

• PostgresSessionStorage: An alternative for production if you are already using PostgreSQL. Requires the postgres extra (pip install gobstopper[postgres]).

Example: Using Redis

from redis.asyncio import Redis
from gobstopper.middleware.security import SecurityMiddleware
from gobstopper.sessions.redis_storage import AsyncRedisSessionStorage

redis_client = Redis.from_url("redis://localhost")
storage = AsyncRedisSessionStorage(client=redis_client)
app.add_middleware(SecurityMiddleware, session_storage=storage)

PostgreSQL Schema

If you use PostgresSessionStorage, you need to create the sessions table in your database.

CREATE TABLE IF NOT EXISTS sessions (
  session_id TEXT PRIMARY KEY,
  data JSONB NOT NULL,
  expires_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_sessions_expires_at ON sessions (expires_at);

Request/Response Lifecycle

Gobstopper provides hooks for request lifecycle management:

Before Request Handlers

@app.before_request
async def authenticate_user(request):
    # Authentication logic
    token = request.headers.get("Authorization")
    if token:
        user = await verify_token(token)
        request.user = user

@app.before_request
def add_request_timestamp(request):
    request.start_time = time.time()

After Request Handlers

@app.after_request
async def add_security_headers(request, response):
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    return response

@app.after_request
def log_request_duration(request, response):
    if hasattr(request, 'start_time'):
        duration = time.time() - request.start_time
        app.logger.info(f"Request completed in {duration:.3f}s")
    return response

Startup Handlers

@app.on_startup
async def initialize_database():
    await database.connect()
    await database.create_tables()

@app.on_startup
async def load_configuration():
    app.config = await load_config_from_file("config.yaml")

@app.on_startup
def register_signal_handlers():
    signal.signal(signal.SIGTERM, graceful_shutdown)

Error Handling

Custom Error Handlers

@app.error_handler(404)
async def not_found(request, error):
    return await app.render_template("errors/404.html", 
                                   path=request.path), 404

@app.error_handler(500)
async def internal_error(request, error):
    app.logger.error(f"Internal error: {error}", exc_info=True)
    
    if app.config.get('DEBUG'):
        return JSONResponse({"error": str(error)}, status=500)
    else:
        return await app.render_template("errors/500.html"), 500

@app.error_handler(403)
async def forbidden(request, error):
    return JSONResponse({"error": "Access denied"}, status=403)

Exception Handling in Routes

@app.get("/users/<user_id>")
async def get_user(request, user_id):
    try:
        user_id = int(user_id)
        user = await get_user_by_id(user_id)
        
        if not user:
            raise HTTPException(404, "User not found")
            
        return {"user": user}
        
    except ValueError:
        raise HTTPException(400, "Invalid user ID format")
    except DatabaseError as e:
        app.logger.error(f"Database error: {e}")
        raise HTTPException(500, "Database error")

Hybrid Architecture

Rust Component Detection

Gobstopper automatically detects and uses Rust components:

# Check what components are available
if app.rust_router_available:
    print("✅ Using Rust router for high performance")
else:
    print("⚠️ Using Python router (fallback)")

# Template engine detection
if isinstance(app.template_engine, RustTemplateEngineWrapper):
    print("🦀 Using Rust template engine")
else:
    print("📄 Using Jinja2 template engine")

Performance Benefits

Component	Python	Rust	Improvement
HTTP Routing	Baseline	10x faster	Route matching
Template Rendering	Baseline	5-10x faster	Template compilation
Static Files	Baseline	5-8x faster	File serving
Memory Usage	Baseline	30-60% less	Optimized data structures

Fallback Behavior

If Rust components are unavailable, Gobstopper seamlessly falls back to Python implementations:

# Application code remains identical regardless of backend
@app.get("/api/users/<user_id>")
async def get_user(request, user_id):
    # Works with both Rust and Python routers
    return {"user_id": user_id}

# Template rendering works with both engines  
@app.get("/")
async def index(request):
    # Uses Rust or Jinja2 transparently
    return await app.render_template("index.html")

Complete Application Example

from gobstopper import Gobstopper, Request
from gobstopper.middleware.cors import CORSMiddleware
from gobstopper.middleware.security import SecurityMiddleware
from gobstopper.tasks.queue import TaskPriority

# Initialize application
app = Gobstopper(__name__)

# Initialize template engine (auto-detects Rust)
app.init_templates("templates/")

# Add middleware
app.add_middleware(SecurityMiddleware, priority=100)
app.add_middleware(CORSMiddleware, priority=90)

# Global template context
@app.context_processor
def inject_globals():
    return {
        'app_name': 'Gobstopper Demo',
        'version': '1.0.0'
    }

# Background task
@app.task("send_notification", category="notifications")
async def send_notification(user_id: int, message: str):
    # Notification logic
    return {"status": "sent", "user_id": user_id}

# HTTP routes
@app.get("/")
async def index(request: Request):
    return await app.render_template("index.html", 
                                   title="Gobstopper Application")

@app.get("/api/users/<user_id>")
async def get_user(request: Request, user_id: str):
    user = await get_user_by_id(int(user_id))
    if not user:
        raise HTTPException(404, "User not found")
    return {"user": user}

@app.post("/api/notify")
async def queue_notification(request: Request):
    data = await request.json()
    
    task_id = await app.add_background_task(
        "send_notification",
        priority=TaskPriority.HIGH,
        user_id=data["user_id"],
        message=data["message"]
    )
    
    return {"task_id": task_id}

# WebSocket endpoint
@app.websocket("/ws/live")
async def live_updates(websocket):
    await websocket.accept()
    
    while True:
        # Send live updates
        data = await get_live_data()
        await websocket.send_json(data)
        await asyncio.sleep(1)

# Error handlers
@app.error_handler(404)
async def not_found(request, error):
    return await app.render_template("404.html"), 404

# Startup initialization
@app.on_startup
async def initialize():
    await app.start_task_workers("notifications", worker_count=2)
    app.logger.info("🚀 Gobstopper application started")

# Run with: granian --interface rsgi app:app

RSGI Protocol Integration

Gobstopper is built specifically for Granian’s RSGI protocol:

Running the Application

# Basic usage
granian --interface rsgi app:app

# With hot reload (development)
granian --interface rsgi --reload app:app

# Production configuration
granian --interface rsgi --workers 4 --host 0.0.0.0 --port 8000 app:app

RSGI Benefits

• Better Performance: 2-3x faster than ASGI

• Lower Memory Usage: Optimized protocol implementation

• Native Support: Direct integration with Gobstopper’s architecture

• WebSocket Efficiency: Superior WebSocket handling

Best Practices

Application Structure

# Organize large applications
from gobstopper import Gobstopper
from .routes import auth, api, admin
from .middleware import setup_middleware
from .tasks import register_tasks

def create_app():
    app = Gobstopper(__name__)
    
    # Initialize components
    app.init_templates("templates/")
    setup_middleware(app)
    register_tasks(app)
    
    # Register blueprints/routes
    auth.register_routes(app)
    api.register_routes(app)  
    admin.register_routes(app)
    
    return app

app = create_app()

Performance Optimization

1. Use Rust Components: Ensure gobstopper_core_rs is installed

2. Enable Template Caching: Set appropriate cache sizes

3. Background Tasks: Offload heavy work to task queue

4. Static Files: Use RustStaticMiddleware for static assets

5. Database Connections: Use connection pooling

Error Handling

1. Comprehensive Coverage: Handle all expected error conditions

2. Logging: Use structured logging with request context

3. User-Friendly Messages: Don’t expose internal errors to users

4. Monitoring: Implement error tracking and alerting

Security

1. Input Validation: Validate all user inputs

2. Authentication: Implement proper authentication

3. HTTPS: Use HTTPS in production

4. Security Headers: Use SecurityMiddleware

5. CORS: Configure CORS appropriately

The Gobstopper application core provides a solid foundation for building high-performance web applications with the simplicity of Flask and the speed of Rust components.