0
# Schema Processing
1
2
Schema processing utilities enable creation of containers from configuration schemas, allowing dynamic provider creation and dependency graph construction from structured data formats like JSON and YAML.
3
4
## Capabilities
5
6
### Schema Processor
7
8
Core class for processing container schemas and creating providers dynamically.
9
10
```python { .api }
11
class SchemaProcessorV1:
12
"""Processes container schemas from dictionaries."""
13
def __init__(self, schema):
14
"""
15
Initialize schema processor.
16
17
Parameters:
18
- schema: Container schema dictionary
19
"""
20
21
def process(self):
22
"""Process the schema and create providers."""
23
24
def get_providers(self):
25
"""
26
Return the created providers.
27
28
Returns:
29
Dictionary of provider name to provider instance mappings
30
"""
31
```
32
33
### Schema Types
34
35
Type aliases for schema structure definitions.
36
37
```python { .api }
38
ContainerSchema = Dict[Any, Any]
39
"""Type alias for container schema dictionary."""
40
41
ProviderSchema = Dict[Any, Any]
42
"""Type alias for provider schema dictionary."""
43
```
44
45
### Schema Structure
46
47
Schemas define containers and their providers in a structured format that can be loaded from configuration files.
48
49
#### Basic Schema Format
50
51
```python
52
schema = {
53
"container": {
54
"provider_name": {
55
"provider": "Factory",
56
"provides": "myapp.services.UserService",
57
"args": [
58
{"provider": "database"}
59
],
60
"kwargs": {
61
"timeout": 30
62
}
63
},
64
"database": {
65
"provider": "Singleton",
66
"provides": "myapp.database.Database",
67
"args": [
68
"postgresql://localhost/mydb"
69
]
70
}
71
}
72
}
73
```
74
75
#### Schema Processing Example
76
77
```python
78
from dependency_injector import schema, containers
79
80
# Define schema
81
container_schema = {
82
"container": {
83
"config": {
84
"provider": "Configuration"
85
},
86
"database": {
87
"provider": "Singleton",
88
"provides": "myapp.database.Database",
89
"args": [
90
{"provider": "config.database.url"}
91
]
92
},
93
"user_service": {
94
"provider": "Factory",
95
"provides": "myapp.services.UserService",
96
"kwargs": {
97
"database": {"provider": "database"}
98
}
99
}
100
}
101
}
102
103
# Process schema
104
processor = schema.SchemaProcessorV1(container_schema)
105
processor.process()
106
providers_dict = processor.get_providers()
107
108
# Create container with processed providers
109
container = containers.DynamicContainer()
110
container.set_providers(**providers_dict)
111
```
112
113
### JSON Schema Integration
114
115
Loading container schemas from JSON files.
116
117
```python
118
import json
119
from dependency_injector import schema, containers
120
121
# Load schema from JSON file
122
with open("container_schema.json", "r") as f:
123
schema_data = json.load(f)
124
125
# Process schema
126
processor = schema.SchemaProcessorV1(schema_data)
127
processor.process()
128
129
# Create container
130
container = containers.DynamicContainer()
131
container.set_providers(**processor.get_providers())
132
```
133
134
Example JSON schema file (`container_schema.json`):
135
136
```json
137
{
138
"container": {
139
"config": {
140
"provider": "Configuration"
141
},
142
"logger": {
143
"provider": "Singleton",
144
"provides": "logging.Logger",
145
"args": ["myapp"]
146
},
147
"database": {
148
"provider": "Singleton",
149
"provides": "myapp.database.Database",
150
"args": [
151
{"provider": "config.database.url"}
152
]
153
},
154
"user_repository": {
155
"provider": "Factory",
156
"provides": "myapp.repositories.UserRepository",
157
"kwargs": {
158
"database": {"provider": "database"},
159
"logger": {"provider": "logger"}
160
}
161
},
162
"user_service": {
163
"provider": "Factory",
164
"provides": "myapp.services.UserService",
165
"kwargs": {
166
"repository": {"provider": "user_repository"}
167
}
168
}
169
}
170
}
171
```
172
173
### YAML Schema Integration
174
175
Loading container schemas from YAML files.
176
177
```python
178
import yaml
179
from dependency_injector import schema, containers
180
181
# Load schema from YAML file
182
with open("container_schema.yaml", "r") as f:
183
schema_data = yaml.safe_load(f)
184
185
# Process schema
186
processor = schema.SchemaProcessorV1(schema_data)
187
processor.process()
188
189
# Create container
190
container = containers.DynamicContainer()
191
container.set_providers(**processor.get_providers())
192
```
193
194
Example YAML schema file (`container_schema.yaml`):
195
196
```yaml
197
container:
198
config:
199
provider: Configuration
200
201
logger:
202
provider: Singleton
203
provides: logging.Logger
204
args:
205
- myapp
206
207
database:
208
provider: Singleton
209
provides: myapp.database.Database
210
args:
211
- provider: config.database.url
212
213
redis:
214
provider: Resource
215
provides: myapp.resources.RedisResource
216
args:
217
- provider: config.redis.host
218
- provider: config.redis.port
219
220
user_repository:
221
provider: Factory
222
provides: myapp.repositories.UserRepository
223
kwargs:
224
database:
225
provider: database
226
logger:
227
provider: logger
228
229
user_service:
230
provider: Factory
231
provides: myapp.services.UserService
232
kwargs:
233
repository:
234
provider: user_repository
235
cache:
236
provider: redis
237
```
238
239
### Provider Type Mapping
240
241
Schema processor supports various provider types that map to dependency injector providers.
242
243
```python
244
# Supported provider types in schemas
245
provider_types = {
246
"Factory": providers.Factory,
247
"Singleton": providers.Singleton,
248
"ThreadSafeSingleton": providers.ThreadSafeSingleton,
249
"ThreadLocalSingleton": providers.ThreadLocalSingleton,
250
"Callable": providers.Callable,
251
"Coroutine": providers.Coroutine,
252
"Configuration": providers.Configuration,
253
"Resource": providers.Resource,
254
"Object": providers.Object,
255
"List": providers.List,
256
"Dict": providers.Dict,
257
"Dependency": providers.Dependency,
258
"Selector": providers.Selector
259
}
260
```
261
262
### Advanced Schema Features
263
264
#### Provider References
265
266
Reference other providers within the schema:
267
268
```yaml
269
container:
270
database:
271
provider: Singleton
272
provides: myapp.database.Database
273
args:
274
- "postgresql://localhost/mydb"
275
276
user_service:
277
provider: Factory
278
provides: myapp.services.UserService
279
kwargs:
280
database:
281
provider: database # Reference to database provider
282
```
283
284
#### Configuration References
285
286
Reference configuration values:
287
288
```yaml
289
container:
290
config:
291
provider: Configuration
292
293
database:
294
provider: Singleton
295
provides: myapp.database.Database
296
args:
297
- provider: config.database.url # Reference config value
298
```
299
300
#### Nested Provider Arguments
301
302
Complex argument structures:
303
304
```yaml
305
container:
306
email_service:
307
provider: Factory
308
provides: myapp.services.EmailService
309
kwargs:
310
smtp_config:
311
provider: Dict
312
kwargs:
313
host:
314
provider: config.smtp.host
315
port:
316
provider: config.smtp.port
317
username:
318
provider: config.smtp.username
319
password:
320
provider: config.smtp.password
321
```
322
323
### Container Integration
324
325
Integrate schema-created providers with declarative containers.
326
327
```python
328
from dependency_injector import containers, providers, schema
329
330
class BaseContainer(containers.DeclarativeContainer):
331
"""Base container with core providers."""
332
config = providers.Configuration()
333
logger = providers.Singleton(logging.Logger, "myapp")
334
335
# Load additional providers from schema
336
schema_data = load_schema_from_file("additional_providers.yaml")
337
processor = schema.SchemaProcessorV1(schema_data)
338
processor.process()
339
additional_providers = processor.get_providers()
340
341
# Extend base container
342
class ApplicationContainer(BaseContainer):
343
pass
344
345
# Add schema-based providers
346
for name, provider in additional_providers.items():
347
setattr(ApplicationContainer, name, provider)
348
349
# Use combined container
350
container = ApplicationContainer()
351
container.config.from_yaml("config.yaml")
352
```
353
354
### Dynamic Provider Creation
355
356
Create providers dynamically based on runtime conditions.
357
358
```python
359
def create_container_from_environment():
360
"""Create container based on environment configuration."""
361
environment = os.getenv("ENVIRONMENT", "development")
362
363
# Load environment-specific schema
364
schema_file = f"schemas/{environment}_schema.yaml"
365
with open(schema_file, "r") as f:
366
schema_data = yaml.safe_load(f)
367
368
# Process schema
369
processor = schema.SchemaProcessorV1(schema_data)
370
processor.process()
371
372
# Create container
373
container = containers.DynamicContainer()
374
container.set_providers(**processor.get_providers())
375
376
return container
377
378
# Usage
379
container = create_container_from_environment()
380
```
381
382
### Schema Validation
383
384
Validate schema structure before processing.
385
386
```python
387
def validate_schema(schema_data):
388
"""Validate schema structure."""
389
required_keys = ["container"]
390
391
if not isinstance(schema_data, dict):
392
raise ValueError("Schema must be a dictionary")
393
394
if "container" not in schema_data:
395
raise ValueError("Schema must contain 'container' key")
396
397
container_def = schema_data["container"]
398
if not isinstance(container_def, dict):
399
raise ValueError("Container definition must be a dictionary")
400
401
for provider_name, provider_def in container_def.items():
402
if not isinstance(provider_def, dict):
403
raise ValueError(f"Provider '{provider_name}' must be a dictionary")
404
405
if "provider" not in provider_def:
406
raise ValueError(f"Provider '{provider_name}' must specify 'provider' type")
407
408
return True
409
410
# Use validation
411
try:
412
validate_schema(schema_data)
413
processor = schema.SchemaProcessorV1(schema_data)
414
processor.process()
415
except ValueError as e:
416
print(f"Schema validation error: {e}")
417
```
418
419
### Error Handling
420
421
Handle errors during schema processing.
422
423
```python
424
def safe_process_schema(schema_data):
425
"""Process schema with error handling."""
426
try:
427
processor = schema.SchemaProcessorV1(schema_data)
428
processor.process()
429
return processor.get_providers()
430
except ImportError as e:
431
print(f"Failed to import provider class: {e}")
432
return {}
433
except Exception as e:
434
print(f"Schema processing error: {e}")
435
return {}
436
437
# Usage
438
providers_dict = safe_process_schema(schema_data)
439
if providers_dict:
440
container.set_providers(**providers_dict)
441
else:
442
print("Using fallback configuration")
443
```