#
Plugin System
Sonora v1.2.8 includes a powerful plugin system that allows you to extend functionality without modifying core code.
#
Table of Contents
Overview Creating Plugins Plugin Lifecycle Event System Security Examples Best Practices
#
Overview
Plugins in Sonora are Python modules that can:
- Add new commands and features
- Hook into audio processing
- Modify queue behavior
- Integrate with external services
- Customize bot behavior
#
Key Features
- Sandboxed execution - Plugins run in isolated environments
- Permission system - Granular access control
- Hot reloading - Load/unload plugins without restarting
- Event-driven - React to bot events
- Dependency management - Handle plugin dependencies
#
Creating Plugins
#
Basic Plugin Structure
# my_plugin.py
from sonora.plugins import Plugin
class MyPlugin(Plugin):
def __init__(self):
super().__init__(
name="my_plugin",
version="1.0.0",
description="My custom plugin",
permissions=["track.read", "player.control"]
)
async def on_load(self):
print("Plugin loaded!")
async def on_enable(self):
print("Plugin enabled!")
async def on_track_start(self, event):
track = event.data['track']
print(f"Now playing: {track.title}")
#
Plugin Registration
# In your bot code
from sonora.plugins import PluginManager
from my_plugin import MyPlugin
manager = PluginManager()
# Register and enable plugin
await manager.register_plugin(MyPlugin())
await manager.enable_plugin("my_plugin")
#
Plugin Lifecycle
#
Lifecycle Methods
class LifecyclePlugin(Plugin):
async def on_load(self):
"""Called when plugin is first loaded"""
# Initialize resources
self.database = await create_database_connection()
print("Plugin loaded and database connected")
async def on_enable(self):
"""Called when plugin is enabled"""
# Start background tasks
self.monitor_task = asyncio.create_task(self.monitor_usage())
print("Plugin enabled and monitoring started")
async def on_disable(self):
"""Called when plugin is disabled"""
# Stop background tasks
if self.monitor_task:
self.monitor_task.cancel()
print("Plugin disabled")
async def on_unload(self):
"""Called when plugin is unloaded"""
# Clean up resources
await self.database.close()
print("Plugin unloaded and resources cleaned")
#
States
- Loaded: Plugin code loaded, not active
- Enabled: Plugin active and running
- Disabled: Plugin loaded but inactive
- Unloaded: Plugin completely removed
#
Event System
#
Built-in Events
from sonora import event_manager, EventType
class EventPlugin(Plugin):
async def on_enable(self):
@event_manager.on(EventType.TRACK_START)
async def on_track_start(event):
track = event.data['track']
player = event.data['player']
guild = event.data['guild']
# Log track plays
await self.log_track_play(track, guild)
@event_manager.on(EventType.QUEUE_EMPTY)
async def on_queue_empty(event):
guild = event.data['guild']
# Auto-add recommended tracks
recommendations = await self.get_recommendations(guild)
for track in recommendations[:3]:
player = await self.get_player(guild)
await player.queue.add(track)
async def log_track_play(self, track, guild):
# Implement logging logic
pass
async def get_recommendations(self, guild):
# Implement recommendation logic
return []
#
Custom Events
from sonora.events import Event
class AnalyticsPlugin(Plugin):
async def on_enable(self):
# Emit custom events
@event_manager.on(EventType.TRACK_END)
async def on_track_end(event):
track = event.data['track']
duration = event.data.get('duration', 0)
# Emit custom analytics event
analytics_event = Event(
type="analytics.track_completed",
data={
"track": track,
"duration": duration,
"timestamp": time.time()
}
)
await event_manager.emit(analytics_event)
#
Security
#
Permission System
Plugins must declare required permissions:
class SecurePlugin(Plugin):
def __init__(self):
super().__init__(
name="secure_plugin",
permissions=[
"track.read", # Read track info
"track.modify", # Modify tracks
"queue.read", # Read queue
"queue.modify", # Modify queue
"player.control", # Control playback
"filters.write", # Apply filters
"autoplay.read", # Read autoplay settings
]
)
async def modify_track(self, track):
# Check permission at runtime
if not self.has_permission("track.modify"):
raise PermissionError("Insufficient permissions")
# Safe to modify
track.title = f"[Modified] {track.title}"
#
Sandboxing
class SandboxedPlugin(Plugin):
async def execute_code(self, user_code):
# Code is automatically sandboxed
try:
result = await self.safe_execute(user_code)
return result
except SecurityError as e:
print(f"Security violation: {e}")
return None
async def safe_execute(self, code):
# Plugin execution is sandboxed by Sonora
# Dangerous operations are blocked
exec_globals = {
'print': print,
'len': len,
# ... safe builtins only
}
exec(code, exec_globals)
#
Examples
#
Analytics Plugin
class AnalyticsPlugin(Plugin):
def __init__(self):
super().__init__(
name="analytics",
permissions=["track.read"]
)
self.stats = {}
async def on_enable(self):
@event_manager.on(EventType.TRACK_START)
async def track_started(event):
track_id = event.data['track'].identifier
if track_id not in self.stats:
self.stats[track_id] = {'plays': 0, 'skips': 0}
self.stats[track_id]['plays'] += 1
@event_manager.on(EventType.TRACK_END)
async def track_ended(event):
if event.data.get('reason') == 'skipped':
track_id = event.data['track'].identifier
if track_id in self.stats:
self.stats[track_id]['skips'] += 1
async def get_popular_tracks(self, limit=10):
sorted_tracks = sorted(
self.stats.items(),
key=lambda x: x[1]['plays'],
reverse=True
)
return sorted_tracks[:limit]
#
Auto-Moderation Plugin
class ModerationPlugin(Plugin):
def __init__(self):
super().__init__(
name="moderation",
permissions=["player.control", "track.read"]
)
self.banned_words = ["banned_word1", "banned_word2"]
async def on_enable(self):
@event_manager.on(EventType.TRACK_START)
async def check_track(event):
track = event.data['track']
player = event.data['player']
# Check for banned content
if self.contains_banned_content(track):
await player.skip()
await self.notify_moderators(track, "Banned content detected")
def contains_banned_content(self, track):
title_lower = track.title.lower()
return any(word in title_lower for word in self.banned_words)
async def notify_moderators(self, track, reason):
# Send notification to moderators
print(f"Moderation: {track.title} - {reason}")
#
Custom Filter Plugin
class CustomFilterPlugin(Plugin):
def __init__(self):
super().__init__(
name="custom_filters",
permissions=["filters.write"]
)
async def on_enable(self):
# Register custom filter
from sonora.filters import register_filter
@register_filter("my_echo")
class EchoFilter:
def __init__(self, delay=0.3, feedback=0.4):
self.delay = delay
self.feedback = feedback
def apply(self, audio_data):
# Apply echo effect
return self.apply_echo(audio_data, self.delay, self.feedback)
def apply_echo(self, data, delay, feedback):
# Echo implementation
# This is a simplified example
return data # Placeholder
async def apply_echo_command(self, ctx, delay: float = 0.3):
"""Apply echo filter"""
player = await sonora.get_player(ctx.guild.id)
await player.set_filter("my_echo", delay=delay)
await ctx.send(f"Echo filter applied with {delay}s delay")
#
Best Practices
#
Plugin Development
Request Minimal Permissions
# Good: Only request what's needed permissions=["track.read"] # Bad: Request everything permissions=["*"]Handle Errors Gracefully
async def risky_operation(self): try: result = await self.external_api_call() return result except Exception as e: self.logger.error(f"Operation failed: {e}") return NoneClean Up Resources
async def on_unload(self): # Always clean up if self.database: await self.database.close() if self.background_task: self.background_task.cancel()Document Your Plugin
class DocumentedPlugin(Plugin): """ This plugin does X, Y, and Z. Permissions required: - track.read: To analyze tracks - player.control: To skip inappropriate content Events handled: - TRACK_START: Analyze content - TRACK_END: Log statistics """ pass
#
Security Considerations
Validate Input
def process_user_input(self, user_input): if not isinstance(user_input, str): raise ValueError("Input must be string") if len(user_input) > 1000: raise ValueError("Input too long") # Sanitize input return user_input.strip()Limit Resource Usage
async def rate_limited_operation(self): if time.time() - self.last_operation < 1.0: raise RateLimitError("Too many requests") self.last_operation = time.time() return await self.actual_operation()Log Appropriately
# Good: Log useful information self.logger.info(f"Plugin {self.name} processed {count} items") # Bad: Log sensitive data self.logger.info(f"User {user_id} requested {sensitive_data}")
#
Performance
Use Async Operations
# Good async def fetch_data(self): async with self.http_client.get(url) as response: return await response.json() # Bad def fetch_data(self): return requests.get(url).json() # BlockingCache Results
async def get_expensive_data(self, key): if key in self.cache: return self.cache[key] result = await self.compute_expensive_data(key) self.cache[key] = result return resultBatch Operations
async def process_batch(self, items): # Process multiple items together results = [] for batch in self.chunk_items(items, 10): batch_results = await self.process_batch_items(batch) results.extend(batch_results) return results
#
Plugin Management
#
CLI Commands
# List installed plugins
sonoractl plugin list
# Enable a plugin
sonoractl plugin enable my_plugin
# Disable a plugin
sonoractl plugin disable my_plugin
# Show plugin info
sonoractl plugin info my_plugin
#
Programmatic Management
from sonora.plugins import PluginManager
manager = PluginManager()
# Load plugin from file
await manager.load_plugin('path/to/plugin.py')
# Enable plugin
await manager.enable_plugin('my_plugin')
# Get plugin instance
plugin = manager.get_plugin('my_plugin')
# Disable and unload
await manager.disable_plugin('my_plugin')
await manager.unload_plugin('my_plugin')
#
Troubleshooting
#
Common Issues
Plugin not loading
- Check file syntax and imports
- Verify required permissions
- Check Sonora version compatibility
Permission denied
- Plugin requesting permissions it doesn't need
- Missing permission declarations
- Runtime permission checks failing
Plugin crashes
- Handle exceptions properly
- Use try/catch blocks
- Log errors for debugging
Memory leaks
- Clean up resources in
on_unload - Cancel background tasks
- Remove event listeners
For more help, check the troubleshooting guide or visit our Discord server.