0
# Plugin System
1
2
The plugin system provides plugin management framework for loading and managing application plugins. It supports plugin discovery, loading, and lifecycle management, enabling modular application architecture through external plugin modules.
3
4
## Capabilities
5
6
### Plugin Handler Interface
7
8
Base interface for plugin management functionality that defines the contract for plugin operations.
9
10
```python { .api }
11
class PluginHandler:
12
"""
13
Plugin handler interface for managing application plugins.
14
15
Provides methods for loading individual plugins, plugin lists,
16
and managing plugin lifecycle and state.
17
"""
18
19
def load_plugin(self, plugin_name: str) -> None:
20
"""
21
Load a single plugin by name.
22
23
Args:
24
plugin_name: Name of plugin to load
25
"""
26
27
def load_plugins(self, plugin_list: List[str]) -> None:
28
"""
29
Load multiple plugins from a list.
30
31
Args:
32
plugin_list: List of plugin names to load
33
"""
34
35
def get_loaded_plugins(self) -> List[str]:
36
"""
37
Get list of currently loaded plugins.
38
39
Returns:
40
List of loaded plugin names
41
"""
42
43
def get_enabled_plugins(self) -> List[str]:
44
"""
45
Get list of enabled plugins.
46
47
Returns:
48
List of enabled plugin names
49
"""
50
51
def get_disabled_plugins(self) -> List[str]:
52
"""
53
Get list of disabled plugins.
54
55
Returns:
56
List of disabled plugin names
57
"""
58
```
59
60
## Usage Examples
61
62
### Basic Plugin System
63
64
```python
65
from cement import App, Controller, ex, init_defaults
66
67
CONFIG = init_defaults('myapp')
68
CONFIG['myapp']['plugin_dirs'] = ['./plugins', '/usr/share/myapp/plugins']
69
CONFIG['myapp']['plugins'] = ['backup', 'monitoring', 'notifications']
70
71
class PluginController(Controller):
72
class Meta:
73
label = 'plugin'
74
stacked_on = 'base'
75
stacked_type = 'nested'
76
77
@ex(help='list loaded plugins')
78
def list(self):
79
"""List all loaded plugins."""
80
loaded = self.app.plugin.get_loaded_plugins()
81
enabled = self.app.plugin.get_enabled_plugins()
82
disabled = self.app.plugin.get_disabled_plugins()
83
84
print('Plugin Status:')
85
print(f' Loaded: {loaded}')
86
print(f' Enabled: {enabled}')
87
print(f' Disabled: {disabled}')
88
89
@ex(
90
help='load plugin',
91
arguments=[
92
(['plugin_name'], {'help': 'name of plugin to load'})
93
]
94
)
95
def load(self):
96
"""Load a specific plugin."""
97
plugin_name = self.app.pargs.plugin_name
98
99
try:
100
self.app.plugin.load_plugin(plugin_name)
101
print(f'Successfully loaded plugin: {plugin_name}')
102
except Exception as e:
103
print(f'Failed to load plugin {plugin_name}: {e}')
104
105
@ex(help='reload all plugins')
106
def reload(self):
107
"""Reload all configured plugins."""
108
plugins = self.app.config.get('myapp', 'plugins').split(',')
109
110
try:
111
self.app.plugin.load_plugins(plugins)
112
print(f'Reloaded plugins: {plugins}')
113
except Exception as e:
114
print(f'Failed to reload plugins: {e}')
115
116
class BaseController(Controller):
117
class Meta:
118
label = 'base'
119
120
class MyApp(App):
121
class Meta:
122
label = 'myapp'
123
base_controller = 'base'
124
config_defaults = CONFIG
125
handlers = [BaseController, PluginController]
126
127
with MyApp() as app:
128
app.run()
129
130
# Usage:
131
# myapp plugin list
132
# myapp plugin load backup
133
# myapp plugin reload
134
```
135
136
### Plugin Directory Structure
137
138
```
139
myapp/
140
├── plugins/
141
│ ├── backup/
142
│ │ ├── __init__.py
143
│ │ └── plugin.py
144
│ ├── monitoring/
145
│ │ ├── __init__.py
146
│ │ └── plugin.py
147
│ └── notifications/
148
│ ├── __init__.py
149
│ └── plugin.py
150
└── myapp.py
151
```
152
153
### Sample Plugin Implementation
154
155
```python
156
# plugins/backup/plugin.py
157
from cement import Controller, ex
158
159
class BackupController(Controller):
160
"""Plugin controller for backup functionality."""
161
162
class Meta:
163
label = 'backup'
164
stacked_on = 'base'
165
stacked_type = 'nested'
166
description = 'Backup and restore operations'
167
168
@ex(
169
help='create backup',
170
arguments=[
171
(['--target'], {
172
'help': 'backup target directory',
173
'default': './backups'
174
}),
175
(['--compress'], {
176
'action': 'store_true',
177
'help': 'compress backup files'
178
})
179
]
180
)
181
def create(self):
182
"""Create application backup."""
183
target = self.app.pargs.target
184
compress = self.app.pargs.compress
185
186
print(f'Creating backup to: {target}')
187
print(f'Compression: {"enabled" if compress else "disabled"}')
188
189
# Backup logic here
190
print('Backup completed successfully')
191
192
@ex(
193
help='restore backup',
194
arguments=[
195
(['backup_file'], {'help': 'backup file to restore'})
196
]
197
)
198
def restore(self):
199
"""Restore from backup file."""
200
backup_file = self.app.pargs.backup_file
201
202
print(f'Restoring from: {backup_file}')
203
204
# Restore logic here
205
print('Restore completed successfully')
206
207
def load(app):
208
"""Plugin load function."""
209
# Register the plugin controller
210
app.handler.register(BackupController)
211
app.log.info('Backup plugin loaded')
212
213
def unload(app):
214
"""Plugin unload function."""
215
app.log.info('Backup plugin unloaded')
216
```
217
218
### Plugin Configuration
219
220
```python
221
# plugins/monitoring/plugin.py
222
from cement import Handler, Controller, ex, init_defaults
223
224
class MonitoringHandler(Handler):
225
"""Handler for system monitoring."""
226
227
class Meta:
228
interface = 'monitoring'
229
label = 'system_monitor'
230
config_defaults = {
231
'check_interval': 60,
232
'alert_threshold': 80,
233
'log_metrics': True
234
}
235
236
def check_system_health(self):
237
"""Check system health metrics."""
238
import psutil
239
240
cpu_percent = psutil.cpu_percent(interval=1)
241
memory = psutil.virtual_memory()
242
disk = psutil.disk_usage('/')
243
244
return {
245
'cpu_usage': cpu_percent,
246
'memory_usage': memory.percent,
247
'disk_usage': (disk.used / disk.total) * 100,
248
'timestamp': time.time()
249
}
250
251
class MonitoringController(Controller):
252
"""Controller for monitoring commands."""
253
254
class Meta:
255
label = 'monitor'
256
stacked_on = 'base'
257
stacked_type = 'nested'
258
259
@ex(help='check system status')
260
def status(self):
261
"""Check current system status."""
262
try:
263
monitor = self.app.handler.get('monitoring', 'system_monitor')()
264
metrics = monitor.check_system_health()
265
266
print('System Status:')
267
print(f' CPU Usage: {metrics["cpu_usage"]:.1f}%')
268
print(f' Memory Usage: {metrics["memory_usage"]:.1f}%')
269
print(f' Disk Usage: {metrics["disk_usage"]:.1f}%')
270
271
except Exception as e:
272
print(f'Failed to get system status: {e}')
273
274
@ex(help='start monitoring service')
275
def start(self):
276
"""Start continuous monitoring."""
277
interval = self.app.config.get('monitoring', 'check_interval')
278
threshold = self.app.config.get('monitoring', 'alert_threshold')
279
280
print(f'Starting monitoring (interval: {interval}s, threshold: {threshold}%)')
281
282
# Monitoring loop would go here
283
print('Monitoring service started')
284
285
def load(app):
286
"""Load monitoring plugin."""
287
from cement import Interface
288
289
# Define monitoring interface if not exists
290
class MonitoringInterface(Interface):
291
class Meta:
292
interface = 'monitoring'
293
294
def check_system_health(self):
295
raise NotImplementedError
296
297
# Register interface and handlers
298
app.interface.define(MonitoringInterface)
299
app.handler.register(MonitoringHandler)
300
app.handler.register(MonitoringController)
301
302
app.log.info('Monitoring plugin loaded')
303
```
304
305
### Plugin with Dependencies
306
307
```python
308
# plugins/notifications/plugin.py
309
from cement import Controller, ex
310
311
class NotificationsController(Controller):
312
"""Controller for notification functionality."""
313
314
class Meta:
315
label = 'notify'
316
stacked_on = 'base'
317
stacked_type = 'nested'
318
319
def __init__(self, **kwargs):
320
super().__init__(**kwargs)
321
322
# Check plugin dependencies
323
self.check_dependencies()
324
325
def check_dependencies(self):
326
"""Check if required dependencies are available."""
327
try:
328
import requests
329
self.has_requests = True
330
except ImportError:
331
self.has_requests = False
332
self.app.log.warning('Notifications plugin: requests library not available')
333
334
@ex(
335
help='send email notification',
336
arguments=[
337
(['recipient'], {'help': 'email recipient'}),
338
(['message'], {'help': 'notification message'})
339
]
340
)
341
def email(self):
342
"""Send email notification."""
343
if not self.has_requests:
344
print('Email notifications require the requests library')
345
return
346
347
recipient = self.app.pargs.recipient
348
message = self.app.pargs.message
349
350
print(f'Sending email to {recipient}: {message}')
351
# Email sending logic here
352
print('Email sent successfully')
353
354
@ex(
355
help='send webhook notification',
356
arguments=[
357
(['url'], {'help': 'webhook URL'}),
358
(['message'], {'help': 'notification message'})
359
]
360
)
361
def webhook(self):
362
"""Send webhook notification."""
363
if not self.has_requests:
364
print('Webhook notifications require the requests library')
365
return
366
367
url = self.app.pargs.url
368
message = self.app.pargs.message
369
370
import requests
371
import json
372
373
try:
374
payload = {'message': message, 'timestamp': time.time()}
375
response = requests.post(url, json=payload)
376
377
if response.status_code == 200:
378
print('Webhook notification sent successfully')
379
else:
380
print(f'Webhook failed with status: {response.status_code}')
381
382
except Exception as e:
383
print(f'Failed to send webhook: {e}')
384
385
def load(app):
386
"""Load notifications plugin."""
387
app.handler.register(NotificationsController)
388
app.log.info('Notifications plugin loaded')
389
390
def get_plugin_info():
391
"""Return plugin information."""
392
return {
393
'name': 'notifications',
394
'version': '1.0.0',
395
'description': 'Email and webhook notifications',
396
'dependencies': ['requests'],
397
'author': 'Plugin Developer'
398
}
399
```
400
401
### Plugin Discovery and Management
402
403
```python
404
from cement import App, Controller, ex
405
import os
406
import importlib.util
407
import json
408
409
class PluginManagerController(Controller):
410
"""Advanced plugin management controller."""
411
412
class Meta:
413
label = 'plugins'
414
stacked_on = 'base'
415
stacked_type = 'nested'
416
417
def discover_plugins(self):
418
"""Discover available plugins in plugin directories."""
419
plugin_dirs = self.app.config.get('myapp', 'plugin_dirs').split(',')
420
discovered = []
421
422
for plugin_dir in plugin_dirs:
423
plugin_dir = plugin_dir.strip()
424
if not os.path.exists(plugin_dir):
425
continue
426
427
for item in os.listdir(plugin_dir):
428
plugin_path = os.path.join(plugin_dir, item)
429
430
if os.path.isdir(plugin_path):
431
plugin_file = os.path.join(plugin_path, 'plugin.py')
432
if os.path.exists(plugin_file):
433
discovered.append({
434
'name': item,
435
'path': plugin_path,
436
'file': plugin_file
437
})
438
439
return discovered
440
441
@ex(help='discover available plugins')
442
def discover(self):
443
"""Discover and list available plugins."""
444
plugins = self.discover_plugins()
445
446
print(f'Discovered {len(plugins)} plugin(s):')
447
for plugin in plugins:
448
print(f' • {plugin["name"]} ({plugin["path"]})')
449
450
# Try to get plugin info
451
try:
452
spec = importlib.util.spec_from_file_location(
453
f'{plugin["name"]}.plugin', plugin['file']
454
)
455
module = importlib.util.module_from_spec(spec)
456
spec.loader.exec_module(module)
457
458
if hasattr(module, 'get_plugin_info'):
459
info = module.get_plugin_info()
460
print(f' Version: {info.get("version", "unknown")}')
461
print(f' Description: {info.get("description", "no description")}')
462
463
deps = info.get('dependencies', [])
464
if deps:
465
print(f' Dependencies: {", ".join(deps)}')
466
467
except Exception as e:
468
print(f' Error loading plugin info: {e}')
469
470
@ex(help='show plugin information')
471
def info(self):
472
"""Show detailed information about loaded plugins."""
473
loaded = self.app.plugin.get_loaded_plugins()
474
475
print('Loaded Plugin Information:')
476
for plugin_name in loaded:
477
print(f'\n{plugin_name}:')
478
print(f' Status: Loaded')
479
480
# Try to get additional plugin information
481
try:
482
# This would depend on how plugins store their metadata
483
print(f' Description: Plugin providing {plugin_name} functionality')
484
except Exception:
485
print(f' Description: No additional information available')
486
487
@ex(
488
help='validate plugin',
489
arguments=[
490
(['plugin_name'], {'help': 'name of plugin to validate'})
491
]
492
)
493
def validate(self):
494
"""Validate a plugin before loading."""
495
plugin_name = self.app.pargs.plugin_name
496
plugins = self.discover_plugins()
497
498
plugin = next((p for p in plugins if p['name'] == plugin_name), None)
499
if not plugin:
500
print(f'Plugin {plugin_name} not found')
501
return
502
503
print(f'Validating plugin: {plugin_name}')
504
505
try:
506
# Load plugin module for validation
507
spec = importlib.util.spec_from_file_location(
508
f'{plugin["name"]}.plugin', plugin['file']
509
)
510
module = importlib.util.module_from_spec(spec)
511
spec.loader.exec_module(module)
512
513
# Check required functions
514
required_functions = ['load']
515
for func in required_functions:
516
if not hasattr(module, func):
517
print(f' ✗ Missing required function: {func}')
518
return
519
else:
520
print(f' ✓ Found required function: {func}')
521
522
# Check optional functions
523
optional_functions = ['unload', 'get_plugin_info']
524
for func in optional_functions:
525
if hasattr(module, func):
526
print(f' ✓ Found optional function: {func}')
527
else:
528
print(f' - Optional function not found: {func}')
529
530
print(f' ✓ Plugin {plugin_name} validation passed')
531
532
except Exception as e:
533
print(f' ✗ Plugin validation failed: {e}')
534
535
class BaseController(Controller):
536
class Meta:
537
label = 'base'
538
539
class AdvancedPluginApp(App):
540
class Meta:
541
label = 'myapp'
542
base_controller = 'base'
543
handlers = [BaseController, PluginManagerController]
544
545
with AdvancedPluginApp() as app:
546
app.run()
547
548
# Usage:
549
# myapp plugins discover
550
# myapp plugins info
551
# myapp plugins validate backup
552
```