Cullinan Extension Registration Quick Start¶

Version: v0.90
Feature: Unified Extension Registration and Discovery Pattern
Author: Plumeink

Quick Start¶

1. Middleware Registration (Recommended)¶

Use the @middleware decorator to automatically register middleware:

from cullinan.middleware import middleware, Middleware

@middleware(priority=100)
class LoggingMiddleware(Middleware):
    """Logging middleware - records all requests"""

    def process_request(self, handler):
        print(f"→ Request: {handler.request.uri}")
        return handler  # Return handler to continue processing

    def process_response(self, handler, response):
        print(f"← Response: {response}")
        return response

2. Priority Control¶

Lower priority numbers execute first:

@middleware(priority=10)   # Executes first
class CorsMiddleware(Middleware):
    pass

@middleware(priority=50)   # Executes second
class AuthMiddleware(Middleware):
    pass

@middleware(priority=100)  # Executes third
class LoggingMiddleware(Middleware):
    pass

Recommended Priority Ranges: - 0-50: Critical middleware (CORS, security checks) - 51-100: Standard middleware (logging, metrics) - 101-200: Application-specific middleware

3. Extension Point Discovery¶

Query available extension points:

from cullinan.extensions import list_extension_points

# Query all extension points
all_points = list_extension_points()

# Query extension points by category
middleware_points = list_extension_points(category='middleware')
lifecycle_points = list_extension_points(category='lifecycle')

# Display extension point info
for point in middleware_points:
    print(f"{point['name']}: {point['description']}")

Example output:

Middleware.process_request: Intercept and process requests before they reach handlers
Middleware.process_response: Intercept and process responses before they are sent

4. Query Registered Middleware¶

from cullinan.middleware import get_middleware_registry

registry = get_middleware_registry()
registered = registry.get_registered_middleware()

for mw in registered:
    print(f"{mw['priority']:3d}: {mw['name']}")

Example output:

 10: CorsMiddleware
 50: AuthMiddleware
100: LoggingMiddleware

Complete Example¶

Scenario: Building an Authenticated API¶

from cullinan import configure, run
from cullinan.middleware import middleware, Middleware
from cullinan.controller import controller, get_api
from cullinan.service import service, Service
from cullinan.core import Inject
from cullinan.params import Path

# 1. Define middleware (auto-sorted by priority)

@middleware(priority=10)
class CorsMiddleware(Middleware):
    """CORS middleware - executes first"""
    def process_response(self, handler, response):
        handler.set_header('Access-Control-Allow-Origin', '*')
        return response


@middleware(priority=50)
class AuthMiddleware(Middleware):
    """Auth middleware - executes after CORS"""
    def process_request(self, handler):
        token = handler.request.headers.get('Authorization')
        if not token and handler.request.path.startswith('/api/'):
            handler.set_status(401)
            handler.finish({"error": "Unauthorized"})
            return None  # Short-circuit, stop processing
        return handler


@middleware(priority=100)
class LoggingMiddleware(Middleware):
    """Logging middleware - executes last"""
    def process_request(self, handler):
        print(f"[{handler.request.method}] {handler.request.path}")
        return handler


# 2. Define services

@service
class UserService(Service):
    def get_user(self, user_id):
        return {"id": user_id, "name": "Alice"}


# 3. Define controllers

@controller(url='/api/users')
class UserController:
    user_service: 'UserService' = Inject()

    @get_api(url='/{user_id}')
    async def get_user(self, user_id: int = Path()):
        return self.user_service.get_user(user_id)


# 4. Start application

if __name__ == '__main__':
    configure(
        port=8080,
        debug=True
    )
    run()

Execution Flow:

Request arrives
  ↓
CorsMiddleware (priority=10)
  ↓
AuthMiddleware (priority=50) 
  ↓
LoggingMiddleware (priority=100)
  ↓
UserController.get_user()
  ↓
LoggingMiddleware (response, reverse order)
  ↓
AuthMiddleware (response, reverse order)
  ↓
CorsMiddleware (response, reverse order)
  ↓
Response sent

Backward Compatibility¶

Manual registration is still supported:

from cullinan.middleware import Middleware, get_middleware_registry

class MyMiddleware(Middleware):
    def process_request(self, handler):
        return handler

# Manual registration
registry = get_middleware_registry()
registry.register(MyMiddleware, priority=100)

Both methods can be used together, and the framework will automatically sort by priority.

Extension Point Categories¶

The framework provides 6 major extension point categories:

Middleware
Middleware.process_request
Middleware.process_response
Lifecycle
Service.on_init
Service.on_startup
Service.on_shutdown
Injection
custom_scope
custom_provider
Routing
custom_handler
Configuration
config_provider
Handler
Custom Tornado Handler

FAQ¶

Q1: What happens when middleware returns None?¶

A: Returning None short-circuits the chain, skipping subsequent middleware and handlers:

def process_request(self, handler):
    if not_authorized:
        handler.set_status(401)
        handler.finish({"error": "Unauthorized"})
        return None  # Short-circuit, stop processing
    return handler  # Continue to next middleware

Q2: How to view middleware execution order?¶

A: Use the registry query:

from cullinan.middleware import get_middleware_registry

registry = get_middleware_registry()
for mw in registry.get_registered_middleware():
    print(f"{mw['priority']}: {mw['name']}")

Q3: Can middleware be registered dynamically?¶

A: Yes, but it's recommended to complete registration before application startup:

# Register before startup
registry = get_middleware_registry()
registry.register(DynamicMiddleware, priority=150)

# Then start the application
run()

Q4: Does middleware affect performance?¶

A: Decorator registration overhead is minimal (~1μs), with no runtime overhead. Middleware performance depends on your implementation logic.

Feedback & Support¶

For questions or suggestions: 1. Check example code: examples/extension_registration_demo.py 2. Submit an Issue or PR to the project repository