Feature: Interceptors (@UseInterceptors, NestInterceptor, CallHandler)

[ItayTheDar]

Overview

PyNest has no mechanism to wrap route handler execution with cross-cutting logic. There is no way to transparently add logging, timing, response caching, response mapping, or request transformation without modifying handler code.

This feature request proposes NestJS-compatible Interceptors — a composable layer that wraps handler execution before AND after it runs.

---

Motivation

Interceptors are the most powerful tool in the NestJS pipeline because they can:

1. Execute code before and after a route handler
2. Transform the result returned by a handler
3. Transform exceptions thrown by a handler
4. Extend basic handler behavior (caching, logging, timing)
5. Completely override the handler (serve from cache, short-circuit)

Without interceptors, developers must either:

• Add logging/timing boilerplate inside every handler
• Write FastAPI middleware that operates on raw ASGI scope (no DI access)
• Duplicate response transformation logic across services

---

Proposed API

NestInterceptor — base interface

```python
from abc import ABC, abstractmethod
from nest.common.interceptors import NestInterceptor, ExecutionContext, CallHandler
from typing import Observable # we use AsyncGenerator as Python's equivalent

class NestInterceptor(ABC):
@AbstractMethod
async def intercept(self, context: ExecutionContext, next: CallHandler) -> any:
...

class CallHandler:
async def handle(self) -> any:
"""Call the next handler in the chain (ultimately the route handler)."""
...
```

@UseInterceptors decorator

```python
from nest.common.decorators import UseInterceptors

@controller('/users')
@UseInterceptors(LoggingInterceptor) # controller-level
class UserController:

@Get('/:id')
@UseInterceptors(CacheInterceptor) # route-level
def get_user(self, id: int):
 ...

```

---

Built-in Interceptors

LoggingInterceptor

```python
from nest.common.interceptors import NestInterceptor, ExecutionContext, CallHandler
import time, logging

class LoggingInterceptor(NestInterceptor):
async def intercept(self, context: ExecutionContext, next: CallHandler) -> any:
req = context.switch_to_http().get_request()
logger.info(f"→ {req.method} {req.url}")
start = time.perf_counter()

 result = await next.handle()

 elapsed = (time.perf_counter() - start) * 1000
 logger.info(f"← {req.method} {req.url} [{elapsed:.1f}ms]")
 return result

```

TimeoutInterceptor

```python
import asyncio

class TimeoutInterceptor(NestInterceptor):
def init(self, timeout_ms: int = 5000):
self.timeout = timeout_ms / 1000

async def intercept(self, context: ExecutionContext, next: CallHandler) -> any:
 try:
 return await asyncio.wait_for(next.handle(), timeout=self.timeout)
 except asyncio.TimeoutError:
 raise RequestTimeoutException()

```

CacheInterceptor

```python
class CacheInterceptor(NestInterceptor):
def init(self, cache_service: CacheService):
self.cache = cache_service

async def intercept(self, context: ExecutionContext, next: CallHandler) -> any:
 key = self._get_cache_key(context)
 cached = await self.cache.get(key)
 if cached:
 return cached

 result = await next.handle()
 await self.cache.set(key, result)
 return result

```

TransformInterceptor — wrap every response in {data: ...}

```python
class TransformInterceptor(NestInterceptor):
async def intercept(self, context: ExecutionContext, next: CallHandler) -> any:
result = await next.handle()
return {"data": result, "statusCode": 200}
```

ExcludeNullInterceptor — strip None values from responses

```python
class ExcludeNullInterceptor(NestInterceptor):
async def intercept(self, context: ExecutionContext, next: CallHandler) -> any:
result = await next.handle()
return self._strip_nulls(result)
```

---

ExecutionContext — rich context object

```python
class ExecutionContext:
def switch_to_http(self) -> HttpArgumentsHost: ...
def get_class(self) -> type: ... # the controller class
def get_handler(self) -> callable: ... # the route method
def get_type(self) -> str: ... # 'http' | 'ws'
```

---

app.use_global_interceptors()

```python
app = PyNestFactory.create(AppModule)
app.use_global_interceptors(LoggingInterceptor(), TransformInterceptor())
```

---

Interceptor Execution Order

For a route with both controller-level and route-level interceptors:

→ Global interceptors (outermost)
 → Controller interceptors
 → Route interceptors (innermost)
 → Route handler
 ← Route interceptors
 ← Controller interceptors
← Global interceptors

Each interceptor wraps the next, forming a true middleware onion.

---

DI Support in Interceptors

Interceptors registered with @UseInterceptors(MyInterceptor) should be instantiated via the DI container, allowing them to declare dependencies:

```python
@Injectable
class AuditInterceptor(NestInterceptor):
def init(self, audit_service: AuditService): # injected!
self.audit = audit_service
```

---

Acceptance Criteria

• NestInterceptor abstract base class in nest/common/interceptors.py
• CallHandler class with async handle() method
• ExecutionContext with switch_to_http(), get_class(), get_handler(), get_type()
• @UseInterceptors(*interceptors) decorator for controller and route scope
• app.use_global_interceptors(*interceptors) API
• Interceptors instantiated via DI container (support for injected dependencies)
• Built-in: LoggingInterceptor, TimeoutInterceptor, TransformInterceptor, ExcludeNullInterceptor
• Correct onion-order execution (global → controller → route)
• Async intercept() methods supported
• Unit tests covering all scopes and execution order
• Documentation page

---

Dependencies

• Feature update readme #1 (Exception Filters) — interceptors may throw exceptions that filters catch
• Feature Add official docs #2 (Lifecycle Hooks) — ExecutionContext shares infrastructure with hooks
• Shares ArgumentsHost / HttpArgumentsHost with Exception Filters

---

Related

• NestJS docs: https://docs.nestjs.com/interceptors