How to Create and Chain Method Interceptors¶

Problem¶

You want to add cross-cutting behaviour -- logging, security checks, caching, retry logic -- to specific methods on your components without modifying their business logic.

Solution¶

1. Create an interceptor component¶

An interceptor is a @component that implements the MethodInterceptor protocol: a single invoke(ctx, call_next) method.

import time
from pico_ioc import component, MethodCtx, MethodInterceptor

@component
class TimingInterceptor:
    """Measures and logs method execution time."""

    def invoke(self, ctx: MethodCtx, call_next) -> object:
        start = time.perf_counter()
        result = call_next(ctx)
        elapsed_ms = (time.perf_counter() - start) * 1000
        print(f"{ctx.cls.__name__}.{ctx.name} took {elapsed_ms:.1f}ms")
        return result

2. Apply the interceptor to methods¶

Use @intercepted_by(InterceptorClass) on the methods you want to intercept:

from pico_ioc import component, intercepted_by

@component
class OrderService:
    @intercepted_by(TimingInterceptor)
    def place_order(self, order_id: str) -> str:
        # business logic
        return f"Order {order_id} placed"

3. Chain multiple interceptors¶

Pass multiple interceptor classes. They execute in the order listed, wrapping around each other:

@component
class SecurityInterceptor:
    def invoke(self, ctx: MethodCtx, call_next):
        if not ctx.kwargs.get("user"):
            raise PermissionError("Unauthenticated")
        return call_next(ctx)

@component
class OrderService:
    @intercepted_by(SecurityInterceptor, TimingInterceptor)
    def place_order(self, order_id: str, user: str = "") -> str:
        return f"Order {order_id} placed by {user}"

Execution order:

SecurityInterceptor.invoke()
  -> TimingInterceptor.invoke()
    -> OrderService.place_order()   (the real method)

4. Async interceptors¶

Interceptors work with both sync and async methods. For async methods, the interceptor chain is awaited automatically:

@component
class RetryInterceptor:
    async def invoke(self, ctx: MethodCtx, call_next):
        for attempt in range(3):
            try:
                result = call_next(ctx)
                if hasattr(result, "__await__"):
                    result = await result
                return result
            except Exception:
                if attempt == 2:
                    raise

Explanation¶

When a component has @intercepted_by on any method, the container wraps the entire component in a UnifiedComponentProxy. On each method call:

1. The proxy detects the interceptor classes from _pico_interceptors_.
2. It resolves each interceptor class from the container (they are components themselves).
3. dispatch_method() builds a chain: each call_next invocation calls the next interceptor, and the final call_next calls the real method.
4. Wrapped methods are cached per scope signature to avoid repeated setup.

flowchart LR
    Caller -->|"place_order()"| Proxy["UnifiedComponentProxy"]
    Proxy --> IC1["SecurityInterceptor.invoke()"]
    IC1 -->|"call_next(ctx)"| IC2["TimingInterceptor.invoke()"]
    IC2 -->|"call_next(ctx)"| Real["OrderService.place_order()"]
    Real -->|result| IC2
    IC2 -->|result| IC1
    IC1 -->|result| Proxy
    Proxy -->|result| Caller

The MethodCtx object¶

Every interceptor receives a MethodCtx with:

Common Pitfalls¶