Nitro Dispatch

pip install nitro-dispatch

from nitro_dispatch import PluginManager, PluginBase, hook

class WelcomePlugin(PluginBase):
    name = "welcome"

    @hook('user.login', priority=100)
    def greet_user(self, data):
        print(f"Welcome, {data['username']}!")
        data['greeted'] = True
        return data

manager = PluginManager()
manager.register(WelcomePlugin)
manager.load_all()

result = manager.trigger('user.login', {'username': 'Alice'})
# Output: Welcome, Alice!

class MyPlugin(PluginBase):
    name = "my_plugin"              # Required: unique identifier
    version = "1.0.0"               # Plugin version
    description = "Does cool stuff" # Human-readable description
    author = "Your Name"            # Plugin author
    dependencies = []               # List of required plugin names

class ValidationPlugin(PluginBase):
    name = "validator"

    @hook('before_save', priority=100, timeout=5.0)
    def validate(self, data):
        if not data.get('email'):
            raise ValueError("Email required")
        return data

class LoggingPlugin(PluginBase):
    name = "logger"

    def on_load(self):
        self.register_hook('before_save', self.log_data, priority=50)

    def log_data(self, data):
        print(f"Saving: {data}")
        return data

# Hook 1 (priority=100)
@hook('process_data', priority=100)
def add_timestamp(self, data):
    data['timestamp'] = datetime.now()
    return data

# Hook 2 (priority=50)
@hook('process_data', priority=50)
def add_id(self, data):
    data['id'] = generate_id()
    return data

# Data flows: original → add_timestamp → add_id → final result

class SecurityPlugin(PluginBase):
    @hook('user.login', priority=100)  # Runs first
    def security_check(self, data):
        return data

class LoggingPlugin(PluginBase):
    @hook('user.login', priority=10)  # Runs last
    def log_login(self, data):
        return data

class AsyncPlugin(PluginBase):
    @hook('data.fetch')
    async def fetch_from_api(self, data):
        result = await aiohttp.get('https://api.example.com')
        data['result'] = await result.json()
        return data

# Trigger async
result = await manager.trigger_async('data.fetch', {})

@hook('process_data', timeout=2.0)  # 2 second timeout
def slow_process(self, data):
    # If this takes > 2s, HookTimeoutError is raised
    time.sleep(5)  # This will timeout!
    return data

class AuditPlugin(PluginBase):
    @hook('user.*')  # Matches user.login, user.logout, etc.
    def audit_user_events(self, data):
        log.info(f"User event: {data}")
        return data

    @hook('db.before_*')  # Matches db.before_save, db.before_delete
    def audit_db_operations(self, data):
        log.info(f"DB operation: {data}")
        return data

# Trigger events
manager.trigger('user.login', {})      # Caught by user.*
manager.trigger('db.before_save', {})  # Caught by db.before_*

from nitro_dispatch import StopPropagation

class ValidationPlugin(PluginBase):
    @hook('process_data', priority=100)
    def validate(self, data):
        if not data.get('valid'):
            raise StopPropagation("Invalid data")
        return data

# Hooks with lower priority won't execute if validation fails

manager = PluginManager()

# Discover plugins from directory
discovered = manager.discover_plugins(
    '~/.myapp/plugins',
    pattern='*_plugin.py',
    recursive=True
)

print(f"Discovered: {discovered}")
manager.load_all()

# Reload a specific plugin
manager.reload('my_plugin')

# The plugin will be unloaded, module reloaded, and loaded again

# Disable a plugin (hooks won't execute)
manager.disable_plugin('optional_plugin')

# Re-enable it
manager.enable_plugin('optional_plugin')

def on_plugin_loaded(data):
    print(f"Plugin loaded: {data['plugin_name']}")
    return data

manager.register_hook(
    PluginManager.EVENT_PLUGIN_LOADED,
    on_plugin_loaded
)

# Built-in events:
# - nitro.plugin.registered
# - nitro.plugin.loaded
# - nitro.plugin.unloaded
# - nitro.plugin.error
# - nitro.app.startup
# - nitro.app.shutdown

manager = PluginManager(log_level='DEBUG')
manager.enable_hook_tracing(True)

# Now all hook executions are logged with timing info
result = manager.trigger('user.login', {})

# Log and continue (default)
manager.set_error_strategy('log_and_continue')

# Stop on first error
manager.set_error_strategy('fail_fast')

# Collect all errors
manager.set_error_strategy('collect_all')

config = {
    'cache': {
        'max_size': 100,
        'ttl': 3600
    }
}
manager = PluginManager(config=config)

class CachePlugin(PluginBase):
    name = "cache"

    def on_load(self):
        max_size = self.get_config('max_size', 50)
        ttl = self.get_config('ttl', 1800)

@hook(event_name, priority=50, timeout=None, async_hook=False)

from nitro_dispatch import (
    NitroPluginError,
    PluginLoadError,
    HookTimeoutError,
    StopPropagation,
)

try:
    manager.load('my_plugin')
except PluginLoadError as e:
    print(f"Failed to load plugin: {e}")

python examples/basic_usage.py

python examples/advanced_usage.py
python examples/advanced_features.py

python examples/discovery_example.py

git clone https://github.com/nitrosh/nitro-dispatch.git
cd nitro-dispatch
pip install -e ".[dev]"

pytest
pytest --cov=nitro_dispatch

black nitro_dispatch tests examples