0
# Plugin System
1

2
Extensible plugin architecture for extracting data from requests and enriching responses. The plugin system provides a clean separation between data extraction logic and context management, with built-in plugins for common use cases.
3

4
## Capabilities
5

6
### Base Plugin Classes
7

8
Foundation classes for creating custom plugins that integrate with the middleware system.
9

10
```python { .api }
11
class Plugin(metaclass=abc.ABCMeta):
12
    """
13
    Base class for building those plugins to extract things from request.
14
    
15
    One plugin should be responsible for extracting one thing.
16
    key: the key that allows to access value in headers
17
    """
18
    
19
    key: str  # Header key or context key for this plugin
20
    
21
    async def extract_value_from_header_by_key(
22
        self, request: Union[Request, HTTPConnection]
23
    ) -> Optional[Any]:
24
        """
25
        Extract value from request headers using plugin's key.
26
        
27
        Parameters:
28
        - request: Request or HTTPConnection object
29
        
30
        Returns:
31
        Optional[Any]: Header value or None if not found
32
        """
33
    
34
    async def process_request(
35
        self, request: Union[Request, HTTPConnection]
36
    ) -> Optional[Any]:
37
        """
38
        Runs always on request.
39
        
40
        Extracts value from header by default.
41
        
42
        Parameters:
43
        - request: Request or HTTPConnection object
44
        
45
        Returns:
46
        Optional[Any]: Processed value for context storage
47
        """
48
    
49
    async def enrich_response(self, arg: Union[Response, Message]) -> None:
50
        """
51
        Runs always on response.
52
        
53
        Does nothing by default.
54
        
55
        Parameters:
56
        - arg: Response object (ContextMiddleware) or Message dict (RawContextMiddleware)
57
        """
58

59
class PluginUUIDBase(Plugin):
60
    """Base class for UUID-based plugins with validation and generation."""
61
    
62
    uuid_functions_mapper = {4: uuid.uuid4}  # Supported UUID versions
63
    
64
    def __init__(
65
        self,
66
        force_new_uuid: bool = False,
67
        version: int = 4,
68
        validate: bool = True,
69
        error_response: Optional[Response] = None
70
    ):
71
        """
72
        Initialize UUID plugin.
73
        
74
        Parameters:
75
        - force_new_uuid: Always generate new UUID, ignore request header
76
        - version: UUID version (currently only 4 supported)
77
        - validate: Validate UUID format if present in request
78
        - error_response: Custom response for validation errors
79
        
80
        Raises:
81
        ConfigurationError: If unsupported UUID version specified
82
        """
83
    
84
    def validate_uuid(self, uuid_to_validate: str) -> None:
85
        """
86
        Validate UUID format.
87
        
88
        Parameters:
89
        - uuid_to_validate: UUID string to validate
90
        
91
        Raises:
92
        WrongUUIDError: If UUID format is invalid
93
        """
94
    
95
    def get_new_uuid(self) -> str:
96
        """
97
        Generate new UUID.
98
        
99
        Returns:
100
        str: New UUID as hex string
101
        """
102
    
103
    async def extract_value_from_header_by_key(
104
        self, request: Union[Request, HTTPConnection]
105
    ) -> Optional[str]:
106
        """
107
        Extract or generate UUID from request.
108
        
109
        Parameters:
110
        - request: Request or HTTPConnection object
111
        
112
        Returns:
113
        Optional[str]: UUID string
114
        
115
        Raises:
116
        WrongUUIDError: If validation enabled and UUID is invalid
117
        """
118
    
119
    async def enrich_response(self, arg: Any) -> None:
120
        """
121
        Add UUID to response headers.
122
        
123
        Parameters:
124
        - arg: Response object or Message dict
125
        """
126
```
127

128
### Built-in Plugins
129

130
Ready-to-use plugins for common header extraction and processing scenarios.
131

132
```python { .api }
133
class RequestIdPlugin(PluginUUIDBase):
134
    """Manages request IDs with X-Request-ID header."""
135
    key = HeaderKeys.request_id  # "X-Request-ID"
136

137
class CorrelationIdPlugin(PluginUUIDBase):
138
    """Manages correlation IDs with X-Correlation-ID header."""
139
    key = HeaderKeys.correlation_id  # "X-Correlation-ID"
140

141
class ApiKeyPlugin(Plugin):
142
    """Extracts API key from X-API-Key header."""
143
    key = HeaderKeys.api_key  # "X-API-Key"
144

145
class UserAgentPlugin(Plugin):
146
    """Extracts User-Agent header."""
147
    key = HeaderKeys.user_agent  # "User-Agent"
148

149
class ForwardedForPlugin(Plugin):
150
    """Extracts X-Forwarded-For header."""
151
    key = HeaderKeys.forwarded_for  # "X-Forwarded-For"
152

153
class DateHeaderPlugin(Plugin):
154
    """Parses Date header in RFC1123 format."""
155
    key = HeaderKeys.date  # "Date"
156
    
157
    def __init__(
158
        self,
159
        *args: Any,
160
        error_response: Optional[Response] = Response(status_code=400)
161
    ) -> None:
162
        """
163
        Initialize date header plugin.
164
        
165
        Parameters:
166
        - error_response: Response to return on date format errors
167
        """
168
    
169
    @staticmethod
170
    def rfc1123_to_dt(s: str) -> datetime.datetime:
171
        """
172
        Convert RFC1123 date string to datetime.
173
        
174
        Parameters:
175
        - s: RFC1123 formatted date string
176
        
177
        Returns:
178
        datetime.datetime: Parsed datetime object
179
        
180
        Raises:
181
        ValueError: If date format is invalid
182
        """
183
    
184
    async def process_request(
185
        self, request: Union[Request, HTTPConnection]
186
    ) -> Optional[datetime.datetime]:
187
        """
188
        Parse Date header to datetime.
189
        
190
        Parameters:
191
        - request: Request or HTTPConnection object
192
        
193
        Returns:
194
        Optional[datetime.datetime]: Parsed date or None if not present
195
        
196
        Raises:
197
        DateFormatError: If date format is invalid
198
        """
199
```
200

201
## Usage Examples
202

203
### Basic Plugin Usage
204

205
```python
206
from starlette_context.middleware import ContextMiddleware
207
from starlette_context.plugins import RequestIdPlugin, UserAgentPlugin
208
from starlette_context import context
209

210
# Setup middleware with plugins
211
app.add_middleware(
212
    ContextMiddleware,
213
    plugins=[
214
        RequestIdPlugin(),
215
        UserAgentPlugin()
216
    ]
217
)
218

219
# Access plugin data in handlers
220
async def my_handler(request):
221
    request_id = context["X-Request-ID"]  # From RequestIdPlugin
222
    user_agent = context["User-Agent"]    # From UserAgentPlugin
223
    return {"request_id": request_id, "user_agent": user_agent}
224
```
225

226
### UUID Plugin Configuration
227

228
```python
229
from starlette_context.plugins import RequestIdPlugin, CorrelationIdPlugin
230
from starlette.responses import JSONResponse
231

232
# Always generate new request ID
233
request_id_plugin = RequestIdPlugin(force_new_uuid=True)
234

235
# Use existing correlation ID or generate new one, with validation
236
correlation_plugin = CorrelationIdPlugin(
237
    validate=True,
238
    error_response=JSONResponse(
239
        {"error": "Invalid correlation ID format"}, 
240
        status_code=400
241
    )
242
)
243

244
app.add_middleware(
245
    ContextMiddleware,
246
    plugins=[request_id_plugin, correlation_plugin]
247
)
248
```
249

250
### Date Header Plugin
251

252
```python
253
from starlette_context.plugins import DateHeaderPlugin
254
from starlette_context import context
255
import datetime
256

257
# Parse RFC1123 date headers
258
date_plugin = DateHeaderPlugin()
259

260
app.add_middleware(ContextMiddleware, plugins=[date_plugin])
261

262
async def handler(request):
263
    date_value = context.get("Date")  # datetime.datetime object or None
264
    if date_value:
265
        formatted_date = date_value.strftime("%Y-%m-%d %H:%M:%S")
266
        return {"parsed_date": formatted_date}
267
    return {"parsed_date": None}
268
```
269

270
### Custom Plugin Development
271

272
```python
273
from starlette_context.plugins import Plugin
274
from starlette_context import context
275
import json
276

277
class CustomHeaderPlugin(Plugin):
278
    key = "X-Custom-Data"
279
    
280
    async def process_request(self, request):
281
        # Extract and process header
282
        raw_value = await self.extract_value_from_header_by_key(request)
283
        if raw_value:
284
            try:
285
                # Parse JSON data
286
                return json.loads(raw_value)
287
            except json.JSONDecodeError:
288
                return {"error": "Invalid JSON in header"}
289
        return None
290
    
291
    async def enrich_response(self, response):
292
        # Add processed data to response
293
        custom_data = context.get(self.key)
294
        if custom_data and hasattr(response, 'headers'):
295
            response.headers["X-Processed-Data"] = json.dumps(custom_data)
296

297
# Use custom plugin
298
app.add_middleware(
299
    ContextMiddleware,
300
    plugins=[CustomHeaderPlugin()]
301
)
302
```
303

304
### Advanced UUID Plugin
305

306
```python
307
from starlette_context.plugins import PluginUUIDBase
308
from starlette_context.header_keys import HeaderKeys
309

310
class TraceIdPlugin(PluginUUIDBase):
311
    key = "X-Trace-ID"
312
    
313
    def __init__(self, **kwargs):
314
        # Always validate trace IDs, generate if missing
315
        super().__init__(
316
            force_new_uuid=False,
317
            validate=True,
318
            **kwargs
319
        )
320
    
321
    async def enrich_response(self, response):
322
        # Always add trace ID to response
323
        await super().enrich_response(response)
324
        
325
        # Add to custom header as well
326
        trace_id = context[self.key]
327
        if hasattr(response, 'headers'):
328
            response.headers["X-Trace-Context"] = f"trace-id={trace_id}"
329

330
app.add_middleware(
331
    ContextMiddleware,
332
    plugins=[TraceIdPlugin()]
333
)
334
```
335

336
### Plugin Error Handling
337

338
```python
339
from starlette_context.plugins import DateHeaderPlugin
340
from starlette_context.errors import DateFormatError
341
from starlette.responses import JSONResponse
342

343
# Custom error response for date parsing
344
error_response = JSONResponse(
345
    {
346
        "error": "Invalid date format",
347
        "expected": "RFC1123 format (e.g., 'Wed, 01 Jan 2020 04:27:12 GMT')"
348
    },
349
    status_code=422
350
)
351

352
date_plugin = DateHeaderPlugin(error_response=error_response)
353

354
app.add_middleware(ContextMiddleware, plugins=[date_plugin])
355
```
356

357
### Multiple Header Plugin
358

359
```python
360
from starlette_context.plugins import Plugin
361

362
class MultiHeaderPlugin(Plugin):
363
    key = "combined_headers"
364
    
365
    def __init__(self, header_keys):
366
        self.header_keys = header_keys
367
    
368
    async def process_request(self, request):
369
        headers = {}
370
        for header_key in self.header_keys:
371
            value = request.headers.get(header_key)
372
            if value:
373
                headers[header_key] = value
374
        return headers if headers else None
375

376
# Extract multiple headers into single context entry
377
multi_plugin = MultiHeaderPlugin([
378
    "X-Forwarded-For",
379
    "X-Real-IP", 
380
    "X-Client-ID"
381
])
382

383
app.add_middleware(ContextMiddleware, plugins=[multi_plugin])
384

385
# Access in handler
386
async def handler(request):
387
    headers = context.get("combined_headers", {})
388
    client_ip = (
389
        headers.get("X-Forwarded-For") or 
390
        headers.get("X-Real-IP") or 
391
        "unknown"
392
    )
393
    return {"client_ip": client_ip}
394
```
395

396
## Plugin Development Guidelines
397

398
### Plugin Responsibilities
399

400
1. **Single Purpose**: Each plugin should handle one specific data extraction task
401
2. **Key Naming**: Use descriptive keys that match header names when appropriate
402
3. **Error Handling**: Provide meaningful error responses for validation failures
403
4. **Performance**: Minimize processing overhead in `process_request`
404
5. **Response Enrichment**: Only modify responses when necessary
405

406
### Plugin Patterns
407

408
```python
409
# Simple header extraction
410
class SimplePlugin(Plugin):
411
    key = "X-My-Header"
412
    # Uses default implementation
413

414
# Header processing
415
class ProcessingPlugin(Plugin):
416
    key = "X-Complex-Header"
417
    
418
    async def process_request(self, request):
419
        raw_value = await self.extract_value_from_header_by_key(request)
420
        return self.process_value(raw_value)
421
    
422
    def process_value(self, value):
423
        # Custom processing logic
424
        pass
425

426
# Response enrichment
427
class EnrichingPlugin(Plugin):
428
    key = "X-Data"
429
    
430
    async def enrich_response(self, response):
431
        data = context.get(self.key)
432
        if data and hasattr(response, 'headers'):
433
            response.headers["X-Processed"] = str(data)
434
```
435

436
### Plugin Testing
437

438
```python
439
import pytest
440
from starlette.requests import Request
441
from starlette_context import request_cycle_context
442

443
async def test_custom_plugin():
444
    plugin = CustomHeaderPlugin()
445
    
446
    # Mock request with header
447
    scope = {
448
        "type": "http",
449
        "headers": [(b"x-custom-data", b'{"key": "value"}')]
450
    }
451
    request = Request(scope)
452
    
453
    # Test plugin processing
454
    result = await plugin.process_request(request)
455
    assert result == {"key": "value"}
456
    
457
    # Test with context
458
    with request_cycle_context({"X-Custom-Data": result}):
459
        # Test response enrichment
460
        response = Response()
461
        await plugin.enrich_response(response)
462
        assert "X-Processed-Data" in response.headers
463
```