Module API Reference

This reference covers the complete API for creating and managing StarStreamer modules, including the base classes, registry system, and lifecycle management.

BaseModule Class

BaseModule

BaseModule()

Bases: ABC

Base class for all StarStreamer modules

Source code in src/modules/base.py

def __init__(self) -> None:
    self.name: str = ""
    self.enabled: bool = True
    self._loaded: bool = False

module_name abstractmethod property

module_name: str

Unique name for this module

on_disable async

on_disable() -> None

Called when module is disabled

Override this to perform cleanup tasks when the module is disabled.

Source code in src/modules/base.py

async def on_disable(self) -> None:
    """Called when module is disabled

    Override this to perform cleanup tasks when the module
    is disabled.
    """
    # Default implementation does nothing
    return

on_enable async

on_enable() -> None

Called when module is enabled

Override this to perform setup tasks when the module is first enabled or re-enabled.

Source code in src/modules/base.py

async def on_enable(self) -> None:
    """Called when module is enabled

    Override this to perform setup tasks when the module
    is first enabled or re-enabled.
    """
    # Default implementation does nothing
    return

on_load async

on_load() -> None

Called when module is first loaded

Override this for one-time initialization tasks.

Source code in src/modules/base.py

async def on_load(self) -> None:
    """Called when module is first loaded

    Override this for one-time initialization tasks.
    """
    # Default implementation does nothing
    return

on_unload async

on_unload() -> None

Called when module is unloaded

Override this for cleanup when module is unloaded.

Source code in src/modules/base.py

async def on_unload(self) -> None:
    """Called when module is unloaded

    Override this for cleanup when module is unloaded.
    """
    # Default implementation does nothing
    return

register_actions abstractmethod async

register_actions() -> None

Import and register all module actions

This method should import all action files to trigger their @on_event decorators to register with the event bus.

Source code in src/modules/base.py

@abstractmethod
async def register_actions(self) -> None:
    """Import and register all module actions

    This method should import all action files to trigger
    their @on_event decorators to register with the event bus.
    """
    pass

All StarStreamer modules must inherit from BaseModule and implement the required abstract methods.

Abstract Methods

module_name: str (Property)

@property
@abstractmethod
def module_name(self) -> str:
    """Unique name for this module"""
    pass

Required Implementation: Every module must return a unique string identifier.

Example:

@property
def module_name(self) -> str:
    return "my_custom_module"

register_actions() (Method)

@abstractmethod
async def register_actions(self) -> None:
    """Import and register all module actions"""
    pass

Purpose: Import action files to trigger their @on_event decorators.

Example:

async def register_actions(self) -> None:
    try:
        # Import triggers decorator registration
        from modules.my_module.actions import handlers  # noqa: F401
        self.logger.info("Actions registered successfully")
    except Exception as e:
        self.logger.error(f"Failed to register actions: {e}")
        raise

Lifecycle Hooks

All lifecycle hooks are optional and have default implementations that do nothing.

on_load()

async def on_load(self) -> None:
    """Called when module is first loaded"""

When Called: Once when the module is first loaded into the registry.

Use Cases: - Database schema initialization - Service setup - Resource allocation

on_unload()

async def on_unload(self) -> None:
    """Called when module is unloaded"""

When Called: When the module is removed from the registry.

Use Cases: - Cleanup resources - Close connections - Save state

on_enable()

async def on_enable(self) -> None:
    """Called when module is enabled"""

When Called: - During initial loading (after on_load()) - When manually enabled after being disabled

Use Cases: - Start background tasks - Enable event handlers - Resume module functionality

on_disable()

async def on_disable(self) -> None:
    """Called when module is disabled"""

When Called: When module is manually disabled (but not unloaded).

Use Cases: - Pause background tasks - Temporarily disable features - Suspend event handling

Properties and Attributes

Instance Attributes

def __init__(self) -> None:
    self.name: str = ""              # Module display name
    self.enabled: bool = True        # Whether module is enabled
    self._loaded: bool = False       # Internal loaded state

Complete Module Example

import logging
from modules.base import BaseModule

class MyCustomModule(BaseModule):
    """Example custom module implementation"""

    def __init__(self) -> None:
        super().__init__()
        self.logger = logging.getLogger(__name__)
        self.background_task = None

    @property
    def module_name(self) -> str:
        return "my_custom_module"

    async def register_actions(self) -> None:
        """Register all action handlers"""
        try:
            from modules.my_custom_module.actions import commands  # noqa: F401
            from modules.my_custom_module.actions import events    # noqa: F401
            self.logger.info("Custom module actions registered")
        except Exception as e:
            self.logger.error(f"Failed to register actions: {e}")
            raise

    async def on_load(self) -> None:
        """One-time initialization"""
        self.logger.info("Custom module loaded - initializing resources")
        # Initialize databases, services, etc.

    async def on_enable(self) -> None:
        """Start module functionality"""
        self.logger.info("Custom module enabled")
        # Start background tasks, enable features

    async def on_disable(self) -> None:
        """Pause module functionality"""
        self.logger.info("Custom module disabled")
        # Pause tasks, disable features

    async def on_unload(self) -> None:
        """Final cleanup"""
        self.logger.info("Custom module unloaded")
        # Cleanup resources, save state

---

ModuleRegistry Class

ModuleRegistry

ModuleRegistry(db: Database)

Registry for managing StarStreamer modules

Source code in src/modules/registry.py

def __init__(self, db: Database) -> None:
    self.db = db
    self.logger = logging.getLogger(__name__)
    self.modules: dict[str, BaseModule] = {}
    self._loaded_modules: list[str] = []
    self._module_classes: dict[str, type[BaseModule]] = {}

    # Discover available modules at initialization
    self._discover_modules()

disable_module async

disable_module(name: str) -> bool

Disable a loaded module

Parameters:

Name	Type	Description	Default
name	str	Module name to disable	required

Returns:

Type	Description
bool	True if module was disabled successfully, False otherwise

Source code in src/modules/registry.py

async def disable_module(self, name: str) -> bool:
    """Disable a loaded module

    Args:
        name: Module name to disable

    Returns:
        True if module was disabled successfully, False otherwise
    """
    try:
        if name not in self.modules:
            self.logger.error(f"Module {name} is not loaded")
            return False

        module = self.modules[name]
        await module.on_disable()
        module.enabled = False

        # Update database
        await self.db.execute("UPDATE modules SET enabled = 0 WHERE name = ?", (name,))
        await self.db.commit()

        self.logger.info(f"Disabled module: {name}")
        return True

    except Exception as e:
        self.logger.error(f"Failed to disable module {name}: {e}")
        return False

enable_module async

enable_module(name: str) -> bool

Enable a loaded module

Parameters:

Name	Type	Description	Default
name	str	Module name to enable	required

Returns:

Type	Description
bool	True if module was enabled successfully, False otherwise

Source code in src/modules/registry.py

async def enable_module(self, name: str) -> bool:
    """Enable a loaded module

    Args:
        name: Module name to enable

    Returns:
        True if module was enabled successfully, False otherwise
    """
    try:
        if name not in self.modules:
            self.logger.error(f"Module {name} is not loaded")
            return False

        module = self.modules[name]
        await module.on_enable()
        module.enabled = True

        # Update database
        await self.db.execute("UPDATE modules SET enabled = 1 WHERE name = ?", (name,))
        await self.db.commit()

        self.logger.info(f"Enabled module: {name}")
        return True

    except Exception as e:
        self.logger.error(f"Failed to enable module {name}: {e}")
        return False

get_enabled_modules async

get_enabled_modules() -> list[str]

Get list of enabled modules from database

Returns:

Type	Description
list[str]	List of enabled module names

Source code in src/modules/registry.py

async def get_enabled_modules(self) -> list[str]:
    """Get list of enabled modules from database

    Returns:
        List of enabled module names
    """
    try:
        rows = await self.db.fetchall("SELECT name FROM modules WHERE enabled = 1")
        return [row[0] for row in rows]
    except Exception as e:
        self.logger.error(f"Failed to get enabled modules: {e}")
        return []

get_loaded_modules async

get_loaded_modules() -> list[str]

Get list of currently loaded modules

Returns:

Type	Description
list[str]	List of loaded module names

Source code in src/modules/registry.py

async def get_loaded_modules(self) -> list[str]:
    """Get list of currently loaded modules

    Returns:
        List of loaded module names
    """
    return self._loaded_modules.copy()

get_module_stats async

get_module_stats() -> dict[str, int]

Get module statistics

Returns:

Type	Description
dict[str, int]	Dictionary with module stats

Source code in src/modules/registry.py

async def get_module_stats(self) -> dict[str, int]:
    """Get module statistics

    Returns:
        Dictionary with module stats
    """
    try:
        total_modules = await self.db.fetchval("SELECT COUNT(*) FROM modules") or 0
        enabled_modules = await self.db.fetchval("SELECT COUNT(*) FROM modules WHERE enabled = 1") or 0
        loaded_modules = len(self._loaded_modules)

        return {
            "total_modules": total_modules,
            "enabled_modules": enabled_modules,
            "loaded_modules": loaded_modules,
        }
    except Exception as e:
        self.logger.error(f"Failed to get module stats: {e}")
        return {"total_modules": 0, "enabled_modules": 0, "loaded_modules": 0}

is_module_enabled async

is_module_enabled(name: str) -> bool

Check if a module is enabled

Parameters:

Name	Type	Description	Default
name	str	Module name to check	required

Returns:

Type	Description
bool	True if module is enabled, False otherwise

Source code in src/modules/registry.py

async def is_module_enabled(self, name: str) -> bool:
    """Check if a module is enabled

    Args:
        name: Module name to check

    Returns:
        True if module is enabled, False otherwise
    """
    try:
        result = await self.db.fetchval("SELECT enabled FROM modules WHERE name = ?", (name,))
        return bool(result) if result is not None else False
    except Exception as e:
        self.logger.error(f"Failed to check if module {name} is enabled: {e}")
        return False

load_all_modules async

load_all_modules() -> dict[str, bool]

Load all discovered modules

Returns:

Type	Description
dict[str, bool]	Dict mapping module names to whether they loaded successfully

Source code in src/modules/registry.py

async def load_all_modules(self) -> dict[str, bool]:
    """Load all discovered modules

    Returns:
        Dict mapping module names to whether they loaded successfully
    """
    results = {}
    for module_name in self._module_classes:
        self.logger.info(f"Loading module: {module_name}")
        results[module_name] = await self.load_module(module_name)

    loaded_count = sum(results.values())
    self.logger.info(f"Loaded {loaded_count}/{len(self._module_classes)} modules successfully")
    return results

load_module async

load_module(name: str) -> bool

Dynamically load and register a module

Parameters:

Name	Type	Description	Default
name	str	Module name to load (e.g., "chat", "rpg")	required

Returns:

Type	Description
bool	True if module was loaded successfully, False otherwise

Source code in src/modules/registry.py

async def load_module(self, name: str) -> bool:
    """Dynamically load and register a module

    Args:
        name: Module name to load (e.g., "chat", "rpg")

    Returns:
        True if module was loaded successfully, False otherwise
    """
    try:
        if name in self.modules:
            self.logger.warning(f"Module {name} is already loaded")
            return True

        # Dynamic loading using discovered modules
        if name not in self._module_classes:
            self.logger.error(f"Unknown module: {name}. Available modules: {list(self._module_classes.keys())}")
            return False

        module_class = self._module_classes[name]
        module = module_class()

        # Load the module
        await module.on_load()
        await module.register_actions()

        # Mark as enabled if successful
        await module.on_enable()

        self.modules[name] = module
        self._loaded_modules.append(name)

        # Update database
        await self.db.execute(
            "INSERT OR REPLACE INTO modules (name, enabled, loaded_at) VALUES (?, 1, CURRENT_TIMESTAMP)", (name,)
        )
        await self.db.commit()

        self.logger.info(f"Successfully loaded module: {name}")
        return True

    except Exception as e:
        self.logger.error(f"Failed to load module {name}: {e}")
        return False

reload_module async

reload_module(name: str) -> bool

Hot reload a module using importlib.reload()

Parameters:

Name	Type	Description	Default
name	str	Module name to reload (e.g., "chat", "alerts")	required

Returns:

Type	Description
bool	True if module was reloaded successfully, False otherwise

Source code in src/modules/registry.py

async def reload_module(self, name: str) -> bool:
    """
    Hot reload a module using importlib.reload()

    Args:
        name: Module name to reload (e.g., "chat", "alerts")

    Returns:
        True if module was reloaded successfully, False otherwise
    """
    try:
        # First, unload the module if it's currently loaded
        if name in self.modules:
            await self.unload_module(name)

        # Use importlib to reload the Python module and its submodules
        await self._reload_python_modules(name)

        # Re-discover the module class (in case the class changed)
        self._discover_modules()

        # Load the reloaded module
        return await self.load_module(name)

    except Exception as e:
        self.logger.error(f"Failed to reload module {name}: {e}")
        return False

unload_module async

unload_module(name: str) -> bool

Unload a module

Parameters:

Name	Type	Description	Default
name	str	Module name to unload	required

Returns:

Type	Description
bool	True if module was unloaded successfully, False otherwise

Source code in src/modules/registry.py

async def unload_module(self, name: str) -> bool:
    """Unload a module

    Args:
        name: Module name to unload

    Returns:
        True if module was unloaded successfully, False otherwise
    """
    try:
        if name not in self.modules:
            self.logger.warning(f"Module {name} is not loaded")
            return True

        module = self.modules[name]

        # Disable and unload
        await module.on_disable()
        await module.on_unload()

        # Remove from registry
        del self.modules[name]
        if name in self._loaded_modules:
            self._loaded_modules.remove(name)

        # Update database
        await self.db.execute("UPDATE modules SET enabled = 0 WHERE name = ?", (name,))
        await self.db.commit()

        self.logger.info(f"Successfully unloaded module: {name}")
        return True

    except Exception as e:
        self.logger.error(f"Failed to unload module {name}: {e}")
        return False

The ModuleRegistry manages the lifecycle of all modules in StarStreamer.

Constructor

def __init__(self, db: Database) -> None:
    """Initialize module registry with database connection"""

Parameters: - db: Database instance for persistence

Module Management Methods

load_module(name: str) -> bool

async def load_module(self, name: str) -> bool:
    """Dynamically load and register a module"""

Process: 1. Import module class dynamically 2. Call on_load() lifecycle hook 3. Call register_actions() to register handlers 4. Call on_enable() to activate module 5. Update database state

Returns: True if successful, False otherwise

unload_module(name: str) -> bool

async def unload_module(self, name: str) -> bool:
    """Unload a module"""

Process: 1. Call on_disable() lifecycle hook 2. Call on_unload() lifecycle hook 3. Remove from registry 4. Update database state

enable_module(name: str) -> bool

async def enable_module(self, name: str) -> bool:
    """Enable a loaded module"""

Use Case: Re-enable a disabled module without full reload.

disable_module(name: str) -> bool

async def disable_module(self, name: str) -> bool:
    """Disable a loaded module"""

Use Case: Temporarily disable module functionality.

reload_module(name: str) -> bool

async def reload_module(self, name: str) -> bool:
    """Hot reload a module using importlib.reload()"""

Process: 1. Unload the module if currently loaded 2. Use importlib.reload() to reload Python modules 3. Re-discover the module class (in case class changed) 4. Load the reloaded module with fresh state

Returns: True if reload successful, False otherwise

Use Cases: - Development mode with --reload flag - Manual module updates without restarting - Applying code changes without losing connections

Example:

# Hot reload the chat module
success = await registry.reload_module("chat")
if success:
    logger.info("Chat module reloaded successfully")
else:
    logger.error("Failed to reload chat module")

load_all_modules() -> dict[str, bool]

async def load_all_modules(self) -> dict[str, bool]:
    """Load all discovered modules"""

Process: 1. Discover available modules in modules directory 2. Load each module that has a valid module.py file 3. Track load success/failure for each module

Returns: Dict mapping module names to load success status

Example:

results = await registry.load_all_modules()
# {"chat": True, "alerts": True, "rpg": False}

Hot Reloading Methods

_reload_python_modules(module_name: str) -> None

async def _reload_python_modules(self, module_name: str) -> None:
    """Reload Python modules using importlib.reload()"""

Internal Method: Used by reload_module() to handle the low-level module reloading.

Process: 1. Find all Python modules with the module prefix (e.g., modules.chat.*) 2. Use importlib.reload() on each found module 3. Handle circular imports and dependency order

Note: This is an internal method and should not be called directly.

_discover_modules() -> None

def _discover_modules(self) -> None:
    """Discover available modules from the modules directory"""

Process: 1. Scan the modules directory for subdirectories 2. Look for module.py files in each subdirectory 3. Import and cache module classes for lazy loading 4. Update internal module class registry

Auto-Discovery Rules: - Directory must contain module.py file - Directory name cannot start with underscore (_) - Module class must follow naming convention: {ModuleName}Module

Query Methods

get_enabled_modules() -> list[str]

async def get_enabled_modules(self) -> list[str]:
    """Get list of enabled modules from database"""

Returns: List of module names that are enabled in the database.

get_loaded_modules() -> list[str]

async def get_loaded_modules(self) -> list[str]:
    """Get list of currently loaded modules"""

Returns: List of module names currently loaded in memory.

is_module_enabled(name: str) -> bool

async def is_module_enabled(self, name: str) -> bool:
    """Check if a module is enabled"""

get_module_stats() -> dict[str, int]

async def get_module_stats(self) -> dict[str, int]:
    """Get module statistics"""

Returns:

{
    "total_modules": 5,      # Total modules in database
    "enabled_modules": 3,    # Enabled modules in database
    "loaded_modules": 2      # Currently loaded in memory
}

Usage Examples

Web Interface Management (Recommended)

StarStreamer v0.6.1+ includes a web interface for module management:

1. Access the Modules Page:
2. Start StarStreamer: uv run python src/main.py

3. Open http://localhost:8888/modules

4. Available Actions:

5. Load - Load an unloaded module into memory
6. Enable - Activate a loaded but disabled module
7. Disable - Deactivate a module while keeping it loaded

8. Reload - Hot reload module code (useful for development)

9. Real-time Statistics:

10. Total modules discovered
11. Currently enabled modules
12. Currently loaded modules
13. Module status indicators

Programmatic Module Management

from modules.registry import ModuleRegistry
from starstreamer.db.database import Database

# Initialize registry
db = Database("data/starstreamer.db")
registry = ModuleRegistry(db)

# Load modules
await registry.load_module("chat")
await registry.load_module("alerts")
await registry.load_module("rpg")

# Check status
enabled = await registry.get_enabled_modules()
loaded = await registry.get_loaded_modules()
stats = await registry.get_module_stats()

print(f"Enabled: {enabled}")
print(f"Loaded: {loaded}")
print(f"Stats: {stats}")

Dynamic Module Control

# Temporarily disable a module
await registry.disable_module("rpg")

# Re-enable it later
await registry.enable_module("rpg")

# Completely unload a module
await registry.unload_module("chat")

# Load it back
await registry.load_module("chat")

---

Module State Management

State Persistence

Module states are persisted in the database:

CREATE TABLE modules (
    name TEXT PRIMARY KEY,
    enabled BOOLEAN DEFAULT 1,
    loaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

State Transitions

stateDiagram-v2
    [*] --> Unloaded
    Unloaded --> Loading: load_module()
    Loading --> Loaded: on_load() success
    Loading --> Unloaded: on_load() failure
    Loaded --> Registering: register_actions()
    Registering --> Enabled: on_enable() success
    Registering --> Loaded: register_actions() failure
    Enabled --> Disabled: disable_module()
    Disabled --> Enabled: enable_module()
    Enabled --> Unloaded: unload_module()
    Disabled --> Unloaded: unload_module()

Error Handling

All registry methods include comprehensive error handling:

• Load Failures: Module remains unloaded, error logged
• Registration Failures: Module cleanup performed automatically
• Lifecycle Hook Failures: Operation aborted, state rolled back
• Database Failures: In-memory state preserved, warning logged

---

Best Practices

Module Design

1. Single Responsibility: Each module should have one clear purpose
2. Fail-Safe Initialization: Handle all possible initialization failures
3. Resource Cleanup: Always implement proper cleanup in lifecycle hooks
4. Error Resilience: Don't let one action failure crash the entire module

Lifecycle Hook Implementation

async def on_load(self) -> None:
    """Safe initialization pattern"""
    try:
        # Initialize critical resources
        await self.setup_database()
        await self.load_configuration()
        self.logger.info("Module loaded successfully")
    except Exception as e:
        self.logger.error(f"Failed to load module: {e}")
        # Cleanup partial initialization
        await self.cleanup()
        raise

async def on_unload(self) -> None:
    """Safe cleanup pattern"""
    try:
        # Stop all background tasks
        if hasattr(self, 'background_task') and self.background_task:
            self.background_task.cancel()

        # Close connections
        await self.close_connections()

        self.logger.info("Module unloaded successfully")
    except Exception as e:
        self.logger.error(f"Error during unload: {e}")
        # Don't re-raise - cleanup should be best-effort

Action Registration

async def register_actions(self) -> None:
    """Safe action registration"""
    try:
        # Import modules in dependency order
        from modules.my_module.actions import core_handlers      # noqa: F401
        from modules.my_module.actions import extended_handlers  # noqa: F401
        from modules.my_module.actions import admin_handlers     # noqa: F401

        self.logger.info("All action handlers registered")
    except ImportError as e:
        self.logger.error(f"Missing action module: {e}")
        raise
    except Exception as e:
        self.logger.error(f"Unexpected error registering actions: {e}")
        raise

See Also

• Module System Concepts
• Core Modules Reference
• Dependency Injection
• Event System