0
# Plugin System
1
2
Advanced plugin system for extending pydantic's validation and schema generation capabilities, allowing custom validation logic and integration with external libraries.
3
4
## Capabilities
5
6
### Plugin Protocol
7
8
Base protocol for creating pydantic plugins that can hook into validation and schema generation.
9
10
```python { .api }
11
class PydanticPluginProtocol:
12
"""
13
Protocol for pydantic plugins.
14
15
Plugins can modify validation behavior and schema generation.
16
"""
17
18
def new_schema_validator(self, schema, schema_type, schema_type_path, schema_kind, config, plugin_settings):
19
"""
20
Modify or replace schema validation.
21
22
Args:
23
schema: The schema being processed
24
schema_type: Type associated with the schema
25
schema_type_path: Path to the schema type
26
schema_kind: Kind of schema ('BaseModel', 'TypeAdapter', etc.)
27
config: Validation configuration
28
plugin_settings: Plugin-specific settings
29
30
Returns:
31
New schema validator or None to use default
32
"""
33
34
def new_schema_serializer(self, schema, schema_type, schema_type_path, schema_kind, config, plugin_settings):
35
"""
36
Modify or replace schema serialization.
37
38
Args:
39
schema: The schema being processed
40
schema_type: Type associated with the schema
41
schema_type_path: Path to the schema type
42
schema_kind: Kind of schema
43
config: Validation configuration
44
plugin_settings: Plugin-specific settings
45
46
Returns:
47
New schema serializer or None to use default
48
"""
49
```
50
51
### Validation Handler Protocols
52
53
Protocols for creating validation handlers that process specific validation events.
54
55
```python { .api }
56
class BaseValidateHandlerProtocol:
57
"""
58
Base protocol for validation handlers.
59
"""
60
61
def __call__(self, source_type, field_name=None, field_value=None, **kwargs):
62
"""
63
Handle validation event.
64
65
Args:
66
source_type: Type of the source object
67
field_name (str): Name of the field being validated
68
field_value: Value of the field being validated
69
**kwargs: Additional validation context
70
71
Returns:
72
Validation result
73
"""
74
75
class ValidatePythonHandlerProtocol(BaseValidateHandlerProtocol):
76
"""
77
Protocol for Python object validation handlers.
78
"""
79
80
def __call__(self, source_type, input_value, **kwargs):
81
"""
82
Handle Python object validation.
83
84
Args:
85
source_type: Expected type
86
input_value: Value to validate
87
**kwargs: Validation context
88
89
Returns:
90
Validated value
91
"""
92
93
class ValidateJsonHandlerProtocol(BaseValidateHandlerProtocol):
94
"""
95
Protocol for JSON validation handlers.
96
"""
97
98
def __call__(self, source_type, input_value, **kwargs):
99
"""
100
Handle JSON validation.
101
102
Args:
103
source_type: Expected type
104
input_value: JSON string or bytes to validate
105
**kwargs: Validation context
106
107
Returns:
108
Validated value
109
"""
110
111
class ValidateStringsHandlerProtocol(BaseValidateHandlerProtocol):
112
"""
113
Protocol for string validation handlers.
114
"""
115
116
def __call__(self, source_type, input_value, **kwargs):
117
"""
118
Handle string-based validation.
119
120
Args:
121
source_type: Expected type
122
input_value: String value to validate
123
**kwargs: Validation context
124
125
Returns:
126
Validated value
127
"""
128
```
129
130
### Schema Type Utilities
131
132
Utility types and functions for working with schema types in plugins.
133
134
```python { .api }
135
class SchemaTypePath:
136
"""Named tuple representing path to a schema type."""
137
138
def __init__(self, module, qualname):
139
"""
140
Initialize schema type path.
141
142
Args:
143
module (str): Module name
144
qualname (str): Qualified name within module
145
"""
146
147
@property
148
def module(self):
149
"""str: Module name"""
150
151
@property
152
def qualname(self):
153
"""str: Qualified name"""
154
155
SchemaKind = str # Type alias for schema kinds
156
NewSchemaReturns = dict # Type alias for new schema return values
157
```
158
159
### Plugin Registration
160
161
Functions for registering and managing plugins.
162
163
```python { .api }
164
def register_plugin(plugin):
165
"""
166
Register a pydantic plugin.
167
168
Args:
169
plugin: Plugin instance implementing PydanticPluginProtocol
170
"""
171
172
def get_plugins():
173
"""
174
Get list of registered plugins.
175
176
Returns:
177
list: List of registered plugin instances
178
"""
179
```
180
181
## Usage Examples
182
183
### Basic Plugin Creation
184
185
```python
186
from pydantic.plugin import PydanticPluginProtocol
187
from pydantic import BaseModel
188
from typing import Any, Dict, Optional
189
190
class LoggingPlugin(PydanticPluginProtocol):
191
"""Plugin that logs validation events."""
192
193
def __init__(self, log_file="validation.log"):
194
self.log_file = log_file
195
196
def new_schema_validator(self, schema, schema_type, schema_type_path, schema_kind, config, plugin_settings):
197
"""Add logging to validation."""
198
original_validator = None
199
200
def logging_validator(input_value, **kwargs):
201
with open(self.log_file, 'a') as f:
202
f.write(f"Validating {schema_type} with value: {input_value}\n")
203
204
if original_validator:
205
return original_validator(input_value, **kwargs)
206
return input_value
207
208
return logging_validator
209
210
# Register the plugin
211
from pydantic.plugin import register_plugin
212
register_plugin(LoggingPlugin())
213
214
# Now all pydantic validations will be logged
215
class User(BaseModel):
216
name: str
217
age: int
218
219
user = User(name="John", age=30) # This will log validation
220
```
221
222
### Custom Validation Plugin
223
224
```python
225
from pydantic.plugin import PydanticPluginProtocol, ValidatePythonHandlerProtocol
226
from pydantic import BaseModel
227
import re
228
229
class EmailValidationPlugin(PydanticPluginProtocol):
230
"""Plugin that provides enhanced email validation."""
231
232
def __init__(self, strict_domains=None):
233
self.strict_domains = strict_domains or []
234
235
def new_schema_validator(self, schema, schema_type, schema_type_path, schema_kind, config, plugin_settings):
236
"""Enhance email validation."""
237
238
if schema_type_path and 'email' in schema_type_path.qualname.lower():
239
def enhanced_email_validator(input_value, **kwargs):
240
# Basic email validation
241
if not re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', input_value):
242
raise ValueError("Invalid email format")
243
244
# Check domain restrictions
245
if self.strict_domains:
246
domain = input_value.split('@')[1]
247
if domain not in self.strict_domains:
248
raise ValueError(f"Email domain must be one of: {self.strict_domains}")
249
250
return input_value.lower() # Normalize to lowercase
251
252
return enhanced_email_validator
253
254
return None # Use default validation
255
256
# Register with domain restrictions
257
email_plugin = EmailValidationPlugin(strict_domains=['company.com', 'organization.org'])
258
register_plugin(email_plugin)
259
260
class Employee(BaseModel):
261
name: str
262
work_email: str # Will use enhanced validation
263
264
# This will use the enhanced email validation
265
employee = Employee(name="John", work_email="john@company.com")
266
```
267
268
### Schema Modification Plugin
269
270
```python
271
from pydantic.plugin import PydanticPluginProtocol
272
from pydantic import BaseModel, Field
273
from typing import Any
274
275
class DefaultsPlugin(PydanticPluginProtocol):
276
"""Plugin that adds default values based on field names."""
277
278
DEFAULT_VALUES = {
279
'created_at': '2023-01-01T00:00:00Z',
280
'updated_at': '2023-01-01T00:00:00Z',
281
'version': 1,
282
'active': True
283
}
284
285
def new_schema_validator(self, schema, schema_type, schema_type_path, schema_kind, config, plugin_settings):
286
"""Add automatic defaults for common field names."""
287
288
if hasattr(schema_type, '__fields__'):
289
for field_name, field_info in schema_type.__fields__.items():
290
if field_name in self.DEFAULT_VALUES and field_info.default is None:
291
field_info.default = self.DEFAULT_VALUES[field_name]
292
293
return None # Use default validation
294
295
register_plugin(DefaultsPlugin())
296
297
class Record(BaseModel):
298
id: int
299
name: str
300
created_at: str = None # Will get automatic default
301
active: bool = None # Will get automatic default
302
303
record = Record(id=1, name="Test")
304
print(record.created_at) # "2023-01-01T00:00:00Z"
305
print(record.active) # True
306
```
307
308
### Third-Party Integration Plugin
309
310
```python
311
from pydantic.plugin import PydanticPluginProtocol
312
from pydantic import BaseModel
313
import json
314
315
class JSONSchemaEnhancerPlugin(PydanticPluginProtocol):
316
"""Plugin that enhances JSON schema generation."""
317
318
def new_schema_serializer(self, schema, schema_type, schema_type_path, schema_kind, config, plugin_settings):
319
"""Add enhanced metadata to JSON schemas."""
320
321
original_serializer = None
322
323
def enhanced_serializer(value, **kwargs):
324
# Get original serialized value
325
if original_serializer:
326
result = original_serializer(value, **kwargs)
327
else:
328
result = value
329
330
# Add metadata if this is schema generation
331
if isinstance(result, dict) and 'type' in result:
332
result['x-generated-by'] = 'pydantic-enhanced'
333
result['x-timestamp'] = '2023-12-25T10:30:00Z'
334
335
# Add examples based on field type
336
if result['type'] == 'string' and 'examples' not in result:
337
result['examples'] = ['example string']
338
elif result['type'] == 'integer' and 'examples' not in result:
339
result['examples'] = [42]
340
341
return result
342
343
return enhanced_serializer
344
345
register_plugin(JSONSchemaEnhancerPlugin())
346
347
class Product(BaseModel):
348
name: str
349
price: int
350
351
# JSON schema will include enhanced metadata
352
schema = Product.model_json_schema()
353
print(schema) # Will include x-generated-by and examples
354
```
355
356
### Validation Handler Plugin
357
358
```python
359
from pydantic.plugin import ValidatePythonHandlerProtocol
360
from pydantic import BaseModel
361
from typing import Any
362
363
class TypeCoercionHandler(ValidatePythonHandlerProtocol):
364
"""Handler that provides aggressive type coercion."""
365
366
def __call__(self, source_type, input_value, **kwargs):
367
"""Coerce types more aggressively."""
368
369
# String to number coercion
370
if source_type == int and isinstance(input_value, str):
371
try:
372
return int(float(input_value)) # Handle "42.0" -> 42
373
except ValueError:
374
pass
375
376
# List to string coercion
377
if source_type == str and isinstance(input_value, list):
378
return ', '.join(str(item) for item in input_value)
379
380
# Default behavior
381
return input_value
382
383
class CoercionPlugin(PydanticPluginProtocol):
384
"""Plugin that enables aggressive type coercion."""
385
386
def __init__(self):
387
self.handler = TypeCoercionHandler()
388
389
def new_schema_validator(self, schema, schema_type, schema_type_path, schema_kind, config, plugin_settings):
390
"""Apply type coercion validation."""
391
392
def coercing_validator(input_value, **kwargs):
393
# Apply coercion
394
coerced_value = self.handler(schema_type, input_value, **kwargs)
395
return coerced_value
396
397
return coercing_validator
398
399
register_plugin(CoercionPlugin())
400
401
class Data(BaseModel):
402
count: int
403
tags: str
404
405
# These will work with aggressive coercion
406
data1 = Data(count="42.5", tags=["python", "pydantic"])
407
print(data1.count) # 42 (from "42.5")
408
print(data1.tags) # "python, pydantic" (from list)
409
```
410
411
### Plugin with Settings
412
413
```python
414
from pydantic.plugin import PydanticPluginProtocol
415
from pydantic import BaseModel
416
from typing import Dict, Any
417
418
class CachingPlugin(PydanticPluginProtocol):
419
"""Plugin that caches validation results."""
420
421
def __init__(self, cache_size=1000):
422
self.cache = {}
423
self.cache_size = cache_size
424
425
def new_schema_validator(self, schema, schema_type, schema_type_path, schema_kind, config, plugin_settings):
426
"""Add caching to validation."""
427
428
# Get cache settings from plugin_settings
429
enabled = plugin_settings.get('cache_enabled', True) if plugin_settings else True
430
431
if not enabled:
432
return None
433
434
def caching_validator(input_value, **kwargs):
435
# Create cache key
436
cache_key = f"{schema_type}:{hash(str(input_value))}"
437
438
# Check cache
439
if cache_key in self.cache:
440
return self.cache[cache_key]
441
442
# Validate and cache result
443
# In real implementation, call original validator
444
result = input_value # Simplified
445
446
# Manage cache size
447
if len(self.cache) >= self.cache_size:
448
# Remove oldest entry (simplified)
449
oldest_key = next(iter(self.cache))
450
del self.cache[oldest_key]
451
452
self.cache[cache_key] = result
453
return result
454
455
return caching_validator
456
457
# Register with settings
458
caching_plugin = CachingPlugin(cache_size=500)
459
register_plugin(caching_plugin)
460
461
class CachedModel(BaseModel):
462
value: str
463
464
class Config:
465
# Plugin-specific settings
466
plugin_settings = {
467
'cache_enabled': True
468
}
469
470
# Repeated validations will be cached
471
model1 = CachedModel(value="test")
472
model2 = CachedModel(value="test") # Retrieved from cache
473
```