0
# Extension System
1
2
The extension system provides framework extension loading and management capabilities. It enables modular functionality through well-defined extension points and automatic discovery, allowing applications to be enhanced with additional features.
3
4
## Capabilities
5
6
### Extension Handler Interface
7
8
Base interface for extension loading functionality that defines the contract for extension operations.
9
10
```python { .api }
11
class ExtensionHandler:
12
"""
13
Extension handler interface for loading and managing framework extensions.
14
15
Provides methods for loading individual extensions and extension lists,
16
enabling modular functionality enhancement.
17
"""
18
19
def load_extension(self, ext_module: str) -> None:
20
"""
21
Load a single extension module.
22
23
Args:
24
ext_module: Extension module name to load
25
"""
26
27
def load_extensions(self, ext_list: List[str]) -> None:
28
"""
29
Load multiple extension modules.
30
31
Args:
32
ext_list: List of extension module names to load
33
"""
34
```
35
36
## Built-in Extensions
37
38
Cement provides many built-in extensions for common functionality:
39
40
- **argparse** - Argument parsing and controller system
41
- **colorlog** - Colorized console logging
42
- **configparser** - Configuration file parsing
43
- **dummy** - Basic/dummy handlers for all interfaces
44
- **generate** - Code generation utilities
45
- **jinja2** - Jinja2 template engine integration
46
- **json** - JSON output formatting
47
- **logging** - Python logging integration
48
- **memcached** - Memcached caching backend
49
- **mustache** - Mustache template engine integration
50
- **print** - Simple print-based output
51
- **redis** - Redis caching and data backend
52
- **tabulate** - Tabular output formatting
53
- **watchdog** - File watching capabilities
54
- **yaml** - YAML configuration and output support
55
56
## Usage Examples
57
58
### Basic Extension Loading
59
60
```python
61
from cement import App, Controller, ex
62
63
class MyApp(App):
64
class Meta:
65
label = 'myapp'
66
# Load extensions at startup
67
extensions = [
68
'colorlog', # Colored logging
69
'jinja2', # Template engine
70
'yaml', # YAML support
71
'redis' # Redis caching
72
]
73
74
with MyApp() as app:
75
app.setup()
76
77
# Extensions are now available
78
app.log.info('Application started with extensions')
79
app.run()
80
```
81
82
### Dynamic Extension Loading
83
84
```python
85
from cement import App, Controller, ex
86
87
class ExtensionController(Controller):
88
class Meta:
89
label = 'ext'
90
stacked_on = 'base'
91
stacked_type = 'nested'
92
93
@ex(
94
help='load extension',
95
arguments=[
96
(['extension'], {'help': 'extension name to load'})
97
]
98
)
99
def load(self):
100
"""Load an extension dynamically."""
101
extension_name = self.app.pargs.extension
102
103
try:
104
self.app.extend(extension_name, self.app)
105
print(f'Successfully loaded extension: {extension_name}')
106
except Exception as e:
107
print(f'Failed to load extension {extension_name}: {e}')
108
109
@ex(help='list loaded extensions')
110
def list(self):
111
"""List currently loaded extensions."""
112
if hasattr(self.app, '_loaded_extensions'):
113
extensions = list(self.app._loaded_extensions.keys())
114
print(f'Loaded extensions: {extensions}')
115
else:
116
print('No extensions loaded')
117
118
class BaseController(Controller):
119
class Meta:
120
label = 'base'
121
122
class MyApp(App):
123
class Meta:
124
label = 'myapp'
125
base_controller = 'base'
126
handlers = [BaseController, ExtensionController]
127
128
with MyApp() as app:
129
app.run()
130
131
# Usage:
132
# myapp ext load colorlog
133
# myapp ext load jinja2
134
# myapp ext list
135
```
136
137
### Conditional Extension Loading
138
139
```python
140
from cement import App, Controller, ex, init_defaults
141
import os
142
143
CONFIG = init_defaults('myapp')
144
CONFIG['myapp']['environment'] = os.getenv('APP_ENV', 'development')
145
CONFIG['myapp']['enable_caching'] = True
146
CONFIG['myapp']['enable_templates'] = True
147
148
class ConditionalApp(App):
149
class Meta:
150
label = 'myapp'
151
config_defaults = CONFIG
152
153
def setup(self):
154
# Load base extensions first
155
base_extensions = ['colorlog', 'yaml']
156
157
# Conditionally load extensions based on configuration
158
extensions_to_load = base_extensions.copy()
159
160
if self.config.get('myapp', 'enable_caching'):
161
if os.getenv('REDIS_URL'):
162
extensions_to_load.append('redis')
163
print('Loading Redis extension (REDIS_URL found)')
164
else:
165
print('Caching enabled but no Redis URL found')
166
167
if self.config.get('myapp', 'enable_templates'):
168
extensions_to_load.append('jinja2')
169
print('Loading Jinja2 extension (templates enabled)')
170
171
env = self.config.get('myapp', 'environment')
172
if env == 'development':
173
extensions_to_load.append('watchdog')
174
print('Loading Watchdog extension (development mode)')
175
176
# Load all determined extensions
177
for ext in extensions_to_load:
178
try:
179
self.extend(ext, self)
180
print(f'✓ Loaded extension: {ext}')
181
except Exception as e:
182
print(f'✗ Failed to load extension {ext}: {e}')
183
184
super().setup()
185
186
class BaseController(Controller):
187
class Meta:
188
label = 'base'
189
190
@ex(help='show loaded extensions')
191
def extensions(self):
192
"""Show information about loaded extensions."""
193
print('Extension Status:')
194
195
# Check commonly used extensions
196
extensions_to_check = ['colorlog', 'yaml', 'redis', 'jinja2', 'watchdog']
197
198
for ext in extensions_to_check:
199
if hasattr(self.app, f'_{ext}_loaded'):
200
print(f' ✓ {ext}: Loaded')
201
else:
202
print(f' ✗ {ext}: Not loaded')
203
204
with ConditionalApp() as app:
205
app.run()
206
207
# Usage with environment variables:
208
# APP_ENV=production REDIS_URL=redis://localhost:6379 python myapp.py extensions
209
```
210
211
### Custom Extension Creation
212
213
```python
214
from cement import App, Handler, Interface
215
from cement.core.mail import MailHandler
216
217
class SlackInterface(Interface):
218
"""Interface for Slack messaging functionality."""
219
220
class Meta:
221
interface = 'slack'
222
223
def send_message(self, channel, message):
224
"""Send message to Slack channel."""
225
raise NotImplementedError
226
227
class SlackHandler(Handler, SlackInterface):
228
"""Handler for sending messages to Slack."""
229
230
class Meta:
231
interface = 'slack'
232
label = 'slack'
233
config_defaults = {
234
'webhook_url': None,
235
'username': 'MyApp Bot',
236
'icon_emoji': ':robot_face:'
237
}
238
239
def send_message(self, channel, message):
240
"""Send message to Slack via webhook."""
241
import requests
242
import json
243
244
webhook_url = self.app.config.get('slack', 'webhook_url')
245
if not webhook_url:
246
raise Exception('Slack webhook URL not configured')
247
248
payload = {
249
'channel': channel,
250
'text': message,
251
'username': self.app.config.get('slack', 'username'),
252
'icon_emoji': self.app.config.get('slack', 'icon_emoji')
253
}
254
255
response = requests.post(webhook_url, data=json.dumps(payload))
256
if response.status_code != 200:
257
raise Exception(f'Slack API error: {response.status_code}')
258
259
print(f'Message sent to #{channel}: {message}')
260
261
def load(app):
262
"""Extension load function."""
263
# Define interface
264
app.interface.define(SlackInterface)
265
266
# Register handler
267
app.handler.register(SlackHandler)
268
269
# Usage in application
270
from cement import Controller, ex, init_defaults
271
272
CONFIG = init_defaults('myapp', 'slack')
273
CONFIG['slack']['webhook_url'] = 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'
274
275
class NotificationController(Controller):
276
class Meta:
277
label = 'notify'
278
stacked_on = 'base'
279
stacked_type = 'nested'
280
281
@ex(
282
help='send Slack notification',
283
arguments=[
284
(['channel'], {'help': 'Slack channel name'}),
285
(['message'], {'help': 'message to send'})
286
]
287
)
288
def slack(self):
289
"""Send notification to Slack."""
290
channel = self.app.pargs.channel
291
message = self.app.pargs.message
292
293
# Get Slack handler
294
slack = self.app.handler.get('slack', 'slack')()
295
slack.send_message(channel, message)
296
297
class BaseController(Controller):
298
class Meta:
299
label = 'base'
300
301
class MyApp(App):
302
class Meta:
303
label = 'myapp'
304
base_controller = 'base'
305
config_defaults = CONFIG
306
handlers = [BaseController, NotificationController]
307
308
# Load custom extension
309
def load_custom_extensions(app):
310
load(app) # Load our custom Slack extension
311
312
with MyApp() as app:
313
load_custom_extensions(app)
314
app.run()
315
316
# Usage:
317
# myapp notify slack general "Deployment completed successfully!"
318
```
319
320
### Extension Configuration
321
322
```python
323
from cement import App, init_defaults
324
325
CONFIG = init_defaults('myapp')
326
327
# Configure extensions
328
CONFIG['log.colorlog'] = {
329
'level': 'DEBUG',
330
'format': '%(log_color)s%(levelname)s%(reset)s - %(message)s',
331
'colors': {
332
'DEBUG': 'cyan',
333
'INFO': 'green',
334
'WARNING': 'yellow',
335
'ERROR': 'red',
336
'CRITICAL': 'bold_red'
337
}
338
}
339
340
CONFIG['template.jinja2'] = {
341
'template_dirs': ['./templates', './shared_templates'],
342
'auto_reload': True,
343
'cache_size': 50
344
}
345
346
CONFIG['cache.redis'] = {
347
'host': 'localhost',
348
'port': 6379,
349
'db': 0,
350
'default_timeout': 300
351
}
352
353
CONFIG['output.json'] = {
354
'indent': 2,
355
'sort_keys': True,
356
'ensure_ascii': False
357
}
358
359
class ConfiguredApp(App):
360
class Meta:
361
label = 'myapp'
362
config_defaults = CONFIG
363
extensions = [
364
'colorlog',
365
'jinja2',
366
'redis',
367
'json'
368
]
369
370
# Use loaded extensions
371
log_handler = 'colorlog'
372
template_handler = 'jinja2'
373
cache_handler = 'redis'
374
output_handler = 'json'
375
376
with ConfiguredApp() as app:
377
app.setup()
378
379
# Extensions are configured and ready
380
app.log.info('Application started with configured extensions')
381
382
# Test cache
383
app.cache.set('test_key', 'test_value', time=60)
384
cached_value = app.cache.get('test_key')
385
app.log.info(f'Cache test: {cached_value}')
386
387
# Test output
388
data = {'message': 'Hello from configured app', 'status': 'success'}
389
output = app.render(data)
390
print(output)
391
392
app.run()
393
```
394
395
### Extension Error Handling
396
397
```python
398
from cement import App, Controller, ex
399
400
class RobustApp(App):
401
class Meta:
402
label = 'myapp'
403
404
def setup(self):
405
# Define required and optional extensions
406
required_extensions = ['colorlog', 'yaml']
407
optional_extensions = ['redis', 'jinja2', 'watchdog']
408
409
# Load required extensions (fail if any fail)
410
for ext in required_extensions:
411
try:
412
self.extend(ext, self)
413
print(f'✓ Required extension loaded: {ext}')
414
except Exception as e:
415
print(f'✗ CRITICAL: Failed to load required extension {ext}: {e}')
416
raise SystemExit(1)
417
418
# Load optional extensions (continue if any fail)
419
for ext in optional_extensions:
420
try:
421
self.extend(ext, self)
422
print(f'✓ Optional extension loaded: {ext}')
423
except Exception as e:
424
print(f'⚠ Warning: Failed to load optional extension {ext}: {e}')
425
print(f' Continuing without {ext} functionality')
426
427
super().setup()
428
429
class BaseController(Controller):
430
class Meta:
431
label = 'base'
432
433
@ex(help='test extension availability')
434
def test_extensions(self):
435
"""Test which extensions are available."""
436
extensions_to_test = {
437
'colorlog': 'Colored logging',
438
'yaml': 'YAML configuration',
439
'redis': 'Redis caching',
440
'jinja2': 'Template rendering',
441
'watchdog': 'File watching'
442
}
443
444
print('Extension Availability Test:')
445
for ext, description in extensions_to_test.items():
446
try:
447
# Try to access extension functionality
448
if ext == 'redis' and hasattr(self.app, 'cache'):
449
self.app.cache.set('test', 'value', time=1)
450
status = '✓ Available'
451
elif ext == 'jinja2' and hasattr(self.app, 'template'):
452
status = '✓ Available'
453
elif ext in ['colorlog', 'yaml']: # Always loaded as required
454
status = '✓ Available'
455
else:
456
status = '? Unknown'
457
except Exception:
458
status = '✗ Not available'
459
460
print(f' {ext:12} ({description:20}): {status}')
461
462
with RobustApp() as app:
463
app.run()
464
465
# Usage:
466
# myapp test-extensions
467
```