plugflow


Nameplugflow JSON
Version 1.0.3 PyPI version JSON
download
home_pagehttps://github.com/keklick1337/plugflow
SummaryA powerful Python plugin system with dynamic loading and hot-reload capabilities.
upload_time2025-08-26 12:31:49
maintainerNone
docs_urlNone
authorVladislav Tislenko
requires_python>=3.8
licenseMIT
keywords plugin plugins dynamic-loading hot-reload extensible plugin-system python-plugins
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # PlugFlow

A powerful and flexible Python plugin system that enables dynamic loading, hot-reloading, and management of plugins in your applications. Build extensible software with ease!

[![Python Version](https://img.shields.io/badge/python-3.8%2B-green.svg)](https://python.org) [![Python Version](https://img.shields.io/badge/python-3.9%2B-lime.svg)](https://python.org) [![Python Version](https://img.shields.io/badge/python-3.10%2B-red.svg)](https://python.org) [![Python Version](https://img.shields.io/badge/python-3.11%2B-blue.svg)](https://python.org) [![Python Version](https://img.shields.io/badge/python-3.12%2B-green.svg)](https://python.org) [![Python Version](https://img.shields.io/badge/python-3.13%2B-lime.svg)](https://python.org)
[![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)

## Features

- **Dynamic Plugin Loading**: Load plugins at runtime from files or directories
- **Hot Reload**: Automatically detect and reload plugin changes during development
- **Event System**: Inter-plugin communication through events and hooks
- **Lifecycle Management**: Complete plugin lifecycle with load/unload hooks
- **Type Safety**: Full type hints support for better development experience
- **Error Isolation**: Plugin errors don't crash your application
- **Package Support**: Support for both single-file and package-style plugins
- **Priority System**: Control plugin loading order with priority settings
- **Context Sharing**: Share application context between plugins
- **Framework Agnostic**: Works with any Python application or framework

## Installation

Install PlugFlow using pip:

```bash
pip install plugflow
```

Or install from source:

```bash
git clone https://github.com/keklick1337/plugflow.git
cd plugflow
pip install -e .
```

## Quick Start

### Basic Usage

```python
from plugflow import PluginManager

# Create plugin manager with plugins directory
manager = PluginManager(plugins_paths=["plugins/"])

# Load all plugins from configured paths
manager.load_all()

# Process a command through plugins
result = manager.handle_message("/hello world")
print(result)  # Output from plugin that handles "hello" command

# Send events to plugins
manager.dispatch_event("user_login", {"user_id": 123})
```

### Creating a Simple Plugin

Create a file `plugins/greeter.py`:

```python
from plugflow import BasePlugin

class GreeterPlugin(BasePlugin):
    name = "greeter"
    version = "1.0.0"
    priority = 10  # Higher priority loads first
    
    def on_load(self, manager):
        """Called when plugin is loaded"""
        print(f"Greeter plugin v{self.version} loaded!")
    
    def handle_command(self, command: str, args: str):
        """Handle commands"""
        if command == "hello":
            return f"Hello, {args}!"
        elif command == "goodbye":
            return f"Goodbye, {args}!"
        return None  # Command not handled
    
    def on_event(self, event: str, data, manager):
        """Handle events from other plugins"""
        if event == "user_login":
            print(f"User {data['user_id']} logged in!")
```

### Package-Style Plugin

Create a directory `plugins/advanced_greeter/`:

```
plugins/advanced_greeter/
├── __init__.py
├── handlers.py
└── utils.py
```

`plugins/advanced_greeter/__init__.py`:
```python
from plugflow import BasePlugin
from .handlers import CommandHandler
from .utils import format_message

class AdvancedGreeterPlugin(BasePlugin):
    name = "advanced_greeter"
    version = "2.0.0"
    
    def __init__(self):
        super().__init__()
        self.handler = CommandHandler()
    
    def handle_command(self, command: str, args: str):
        if command in ["greet", "welcome"]:
            return self.handler.handle_greeting(command, args)
        return None
```

## Core Components

### PluginManager

The main class that manages plugin lifecycle and coordination:

```python
from plugflow import PluginManager

manager = PluginManager(
    hot_reload=True,    # Enable hot reload
    context={"app": my_app}  # Share context with plugins
)

# Load plugins
manager.load_from_path(Path("path/to/plugin.py"))
manager.load_all()  # Load all plugins from configured paths

# Manage plugins
manager.unload_plugin("plugin_name")
manager.reload_plugin("plugin_name")

# Plugin communication
result = manager.handle_message("/command args")  # Handle chat-style messages
manager.dispatch_event("event_name", data)

# Get plugin information
plugins = manager.list_plugins()
plugin = manager.get("plugin_name")
```

### BasePlugin

Base class for all plugins:

```python
from plugflow import BasePlugin
from typing import Optional, Any

class MyPlugin(BasePlugin):
    name = "my_plugin"           # Required: unique plugin name
    version = "1.0.0"           # Required: plugin version
    priority = 10               # Optional: loading priority (higher first)
    dependencies = ["other"]    # Optional: plugin dependencies
    
    def on_load(self, manager) -> None:
        """Called when plugin is loaded"""
        pass
    
    def on_unload(self, manager) -> None:
        """Called when plugin is unloaded"""
        pass
    
    def handle_command(self, command: str, args: str) -> Optional[str]:
        """Handle commands - return result or None if not handled"""
        return None
    
    def filter_message(self, text: str) -> Optional[str]:
        """Filter/modify messages - return modified text or None"""
        return None
    
    def on_event(self, event: str, data: Any, manager) -> None:
        """Handle events from other plugins"""
        pass
```

## Advanced Features

### Hot Reload

Enable automatic plugin reloading during development:

```python
manager = PluginManager(
    plugins_paths=["plugins/"],
    hot_reload=True  # Enable hot reload for development
)
manager.load_all()

# Now edit your plugins - changes will be detected automatically!
```

### Context Sharing

Share application state and resources with plugins:

```python
# Create manager with shared context
context = {
    "database": db_connection,
    "config": app_config,
    "logger": logger
}
manager = PluginManager(context=context)

# Plugins can access context
class DatabasePlugin(BasePlugin):
    def on_load(self, manager):
        db = self.context.get("database")
        logger = self.context.get("logger")
        logger.info("Database plugin connected!")
```

### Event System

Plugins can communicate through events:

```python
# Plugin A sends event
class PublisherPlugin(BasePlugin):
    def handle_command(self, command, args):
        if command == "notify":
            # Dispatch event to all plugins
            self.manager.dispatch_event("notification", {
                "message": args,
                "timestamp": time.time()
            })

# Plugin B receives event
class SubscriberPlugin(BasePlugin):
    def on_event(self, event, data, manager):
        if event == "notification":
            print(f"Received: {data['message']}")
```

### Plugin Dependencies

Specify plugin loading order with dependencies:

```python
class DatabasePlugin(BasePlugin):
    name = "database"
    priority = 100  # Load first

class UserPlugin(BasePlugin):
    name = "user_manager"
    dependencies = ["database"]  # Load after database
    
    def on_load(self, manager):
        # Database plugin is guaranteed to be loaded
        db_plugin = manager.get_plugin("database")
```

### Error Handling

Plugins are isolated - errors don't crash your application:

```python
class FaultyPlugin(BasePlugin):
    def handle_command(self, command, args):
        if command == "crash":
            raise Exception("Plugin error!")
        return None

# Manager handles plugin errors gracefully
try:
    result = manager.handle_command("crash", "")
except Exception as e:
    print(f"Plugin error handled: {e}")
# Application continues running
```

## Example Applications

PlugFlow includes comprehensive examples in the `examples/` directory:

### 1. CLI Tool (`examples/cli_tool/`)

Professional command-line utility with plugin-based commands:

```bash
cd examples/cli_tool
python cli.py help          # Show all commands
python cli.py hash md5 "test"  # Cryptographic operations
python cli.py tree . 2      # File system utilities
```

**Features**: Type-safe plugins, clean output, debug mode, comprehensive help

### 2. Telegram Bot (`examples/tg_stub/`)

Production-ready bot simulation with advanced features:

```bash
cd examples/tg_stub
python bot.py              # Clean production mode
python bot.py --debug      # Development mode with logs
```

**Features**: Command handling, message filtering, dynamic help, hot reload

### 3. GUI Application (`examples/tk_app/`)

Sophisticated tkinter application with plugin integration:

```bash
cd examples/tk_app
python app.py
```

**Features**: Menu integration, real-time logging, file operations, event system

### 4. Web Server (`examples/web_server/`)

Flask-based web application with plugin routes:

```bash
cd examples/web_server
python server.py
```

**Features**: Dynamic routing, middleware, API endpoints, template system

## Plugin Development Guide

### Basic Plugin Structure

```python
from plugflow import BasePlugin
from typing import Optional

class MyPlugin(BasePlugin):
    # Plugin metadata
    name = "my_plugin"
    version = "1.0.0"
    description = "My awesome plugin"
    author = "Your Name"
    
    def on_load(self, manager):
        """Initialize plugin resources"""
        self.data = {}
        print(f"{self.name} loaded!")
    
    def on_unload(self, manager):
        """Clean up plugin resources"""
        self.data.clear()
        print(f"{self.name} unloaded!")
    
    def handle_command(self, command: str, args: str) -> Optional[str]:
        """Process commands"""
        commands = {
            "status": self._status,
            "set": self._set_data,
            "get": self._get_data
        }
        
        if command in commands:
            return commands[command](args)
        return None
    
    def _status(self, args: str) -> str:
        return f"Plugin {self.name} v{self.version} - {len(self.data)} items"
    
    def _set_data(self, args: str) -> str:
        key, value = args.split("=", 1)
        self.data[key] = value
        return f"Set {key} = {value}"
    
    def _get_data(self, args: str) -> str:
        return self.data.get(args, "Key not found")
```

### Best Practices

1. **Error Handling**: Always handle exceptions gracefully
2. **Resource Cleanup**: Implement `on_unload` for proper cleanup
3. **Type Hints**: Use type annotations for better code quality
4. **Documentation**: Document your plugin's commands and features
5. **Testing**: Create unit tests for your plugins
6. **Versioning**: Use semantic versioning for your plugins

### Plugin Testing

```python
import unittest
from plugflow import PluginManager
from my_plugin import MyPlugin

class TestMyPlugin(unittest.TestCase):
    def setUp(self):
        self.manager = PluginManager()
        # Manually add plugin for testing
        from plugflow.manager import PluginRecord
        self.plugin = MyPlugin()
        record = PluginRecord(self.plugin, Path("test"), None)
        self.manager._records[self.plugin.name] = record
    
    def test_command_handling(self):
        result = self.manager.handle_message("/status")
        self.assertTrue(any("Plugin my_plugin" in r for r in result))
    
    def test_data_operations(self):
        self.manager.handle_message("/set key=value")
        result = self.manager.handle_message("/get key")
        self.assertTrue(any("value" in r for r in result))
```

## API Reference

### PluginManager

#### Methods

- `load_from_path(path: Path) -> None`: Load plugins from a file or directory
- `load_all() -> None`: Load all plugins from configured paths
- `unload_plugin(name: str) -> bool`: Unload a plugin by name
- `reload_plugin(name: str) -> bool`: Reload a plugin by name
- `handle_message(text: str) -> List[str]`: Process message through plugins (supports /commands and filters)
- `dispatch_event(event: str, data: Any = None) -> List[Any]`: Send event to all plugins
- `broadcast(method: str, *args, **kwargs) -> List[Any]`: Call method on all plugins that have it
- `list_plugins() -> List[str]`: Get list of loaded plugin names
- `get(name: str) -> Optional[BasePlugin]`: Get plugin instance by name
- `stop() -> None`: Stop hot reload watchers

#### Properties

- `hot_reload: bool`: Enable/disable hot reload
- `context: Dict[str, Any]`: Shared context dictionary

### BasePlugin

#### Required Attributes

- `name: str`: Unique plugin identifier
- `version: str`: Plugin version

#### Optional Attributes

- `priority: int`: Loading priority (default: 0)
- `dependencies: List[str]`: Required plugins
- `description: str`: Plugin description
- `author: str`: Plugin author

#### Methods

- `on_load(manager: PluginManager) -> None`: Initialization hook
- `on_unload(manager: PluginManager) -> None`: Cleanup hook
- `handle_command(command: str, args: str) -> Optional[str]`: Command handler
- `filter_message(text: str) -> Optional[str]`: Message filter
- `on_event(event: str, data: Any, manager: PluginManager) -> None`: Event handler

## Performance Tips

1. **Lazy Loading**: Load plugins only when needed
2. **Caching**: Cache plugin results for expensive operations
3. **Priority Optimization**: Use priorities to control loading order
4. **Resource Management**: Properly clean up plugin resources
5. **Event Filtering**: Only subscribe to relevant events

## Troubleshooting

### Common Issues

**Plugin Not Loading**
```python
# Check plugin file syntax
manager.load_from_path(Path("plugin.py"))

# Enable debug logging
import logging
logging.basicConfig(level=logging.DEBUG)
```

**Import Errors**
```python
# Ensure plugin directory is in Python path
import sys
sys.path.append("plugins")
```

**Hot Reload Not Working**
```python
# Ensure hot_reload is enabled
manager = PluginManager(hot_reload=True)

# Check file permissions and watching capability
```

### Debug Mode

Enable verbose logging:

```python
import logging
logging.basicConfig(level=logging.DEBUG)

manager = PluginManager(hot_reload=True)
# Now you'll see detailed plugin loading information
```

## Contributing

We welcome contributions!

### Development Setup

```bash
git clone https://github.com/keklick1337/plugflow.git
cd plugflow
pip install -e ".[dev]"
pre-commit install
```

### Running Tests

```bash
pytest tests/
```

## License

This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

---

**Built with love for the Python community**

PlugFlow makes it easy to create extensible applications. Start building your plugin ecosystem today!

            

Raw data

            {
    "_id": null,
    "home_page": "https://github.com/keklick1337/plugflow",
    "name": "plugflow",
    "maintainer": null,
    "docs_url": null,
    "requires_python": ">=3.8",
    "maintainer_email": null,
    "keywords": "plugin, plugins, dynamic-loading, hot-reload, extensible, plugin-system, python-plugins",
    "author": "Vladislav Tislenko",
    "author_email": "python@trustcrypt.com",
    "download_url": "https://files.pythonhosted.org/packages/3e/f7/777322f47868709f18f33dcdf0ddeed2b124d8b83750867dfb96c1671d13/plugflow-1.0.3.tar.gz",
    "platform": null,
    "description": "# PlugFlow\n\nA powerful and flexible Python plugin system that enables dynamic loading, hot-reloading, and management of plugins in your applications. Build extensible software with ease!\n\n[![Python Version](https://img.shields.io/badge/python-3.8%2B-green.svg)](https://python.org) [![Python Version](https://img.shields.io/badge/python-3.9%2B-lime.svg)](https://python.org) [![Python Version](https://img.shields.io/badge/python-3.10%2B-red.svg)](https://python.org) [![Python Version](https://img.shields.io/badge/python-3.11%2B-blue.svg)](https://python.org) [![Python Version](https://img.shields.io/badge/python-3.12%2B-green.svg)](https://python.org) [![Python Version](https://img.shields.io/badge/python-3.13%2B-lime.svg)](https://python.org)\n[![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)\n\n## Features\n\n- **Dynamic Plugin Loading**: Load plugins at runtime from files or directories\n- **Hot Reload**: Automatically detect and reload plugin changes during development\n- **Event System**: Inter-plugin communication through events and hooks\n- **Lifecycle Management**: Complete plugin lifecycle with load/unload hooks\n- **Type Safety**: Full type hints support for better development experience\n- **Error Isolation**: Plugin errors don't crash your application\n- **Package Support**: Support for both single-file and package-style plugins\n- **Priority System**: Control plugin loading order with priority settings\n- **Context Sharing**: Share application context between plugins\n- **Framework Agnostic**: Works with any Python application or framework\n\n## Installation\n\nInstall PlugFlow using pip:\n\n```bash\npip install plugflow\n```\n\nOr install from source:\n\n```bash\ngit clone https://github.com/keklick1337/plugflow.git\ncd plugflow\npip install -e .\n```\n\n## Quick Start\n\n### Basic Usage\n\n```python\nfrom plugflow import PluginManager\n\n# Create plugin manager with plugins directory\nmanager = PluginManager(plugins_paths=[\"plugins/\"])\n\n# Load all plugins from configured paths\nmanager.load_all()\n\n# Process a command through plugins\nresult = manager.handle_message(\"/hello world\")\nprint(result)  # Output from plugin that handles \"hello\" command\n\n# Send events to plugins\nmanager.dispatch_event(\"user_login\", {\"user_id\": 123})\n```\n\n### Creating a Simple Plugin\n\nCreate a file `plugins/greeter.py`:\n\n```python\nfrom plugflow import BasePlugin\n\nclass GreeterPlugin(BasePlugin):\n    name = \"greeter\"\n    version = \"1.0.0\"\n    priority = 10  # Higher priority loads first\n    \n    def on_load(self, manager):\n        \"\"\"Called when plugin is loaded\"\"\"\n        print(f\"Greeter plugin v{self.version} loaded!\")\n    \n    def handle_command(self, command: str, args: str):\n        \"\"\"Handle commands\"\"\"\n        if command == \"hello\":\n            return f\"Hello, {args}!\"\n        elif command == \"goodbye\":\n            return f\"Goodbye, {args}!\"\n        return None  # Command not handled\n    \n    def on_event(self, event: str, data, manager):\n        \"\"\"Handle events from other plugins\"\"\"\n        if event == \"user_login\":\n            print(f\"User {data['user_id']} logged in!\")\n```\n\n### Package-Style Plugin\n\nCreate a directory `plugins/advanced_greeter/`:\n\n```\nplugins/advanced_greeter/\n\u251c\u2500\u2500 __init__.py\n\u251c\u2500\u2500 handlers.py\n\u2514\u2500\u2500 utils.py\n```\n\n`plugins/advanced_greeter/__init__.py`:\n```python\nfrom plugflow import BasePlugin\nfrom .handlers import CommandHandler\nfrom .utils import format_message\n\nclass AdvancedGreeterPlugin(BasePlugin):\n    name = \"advanced_greeter\"\n    version = \"2.0.0\"\n    \n    def __init__(self):\n        super().__init__()\n        self.handler = CommandHandler()\n    \n    def handle_command(self, command: str, args: str):\n        if command in [\"greet\", \"welcome\"]:\n            return self.handler.handle_greeting(command, args)\n        return None\n```\n\n## Core Components\n\n### PluginManager\n\nThe main class that manages plugin lifecycle and coordination:\n\n```python\nfrom plugflow import PluginManager\n\nmanager = PluginManager(\n    hot_reload=True,    # Enable hot reload\n    context={\"app\": my_app}  # Share context with plugins\n)\n\n# Load plugins\nmanager.load_from_path(Path(\"path/to/plugin.py\"))\nmanager.load_all()  # Load all plugins from configured paths\n\n# Manage plugins\nmanager.unload_plugin(\"plugin_name\")\nmanager.reload_plugin(\"plugin_name\")\n\n# Plugin communication\nresult = manager.handle_message(\"/command args\")  # Handle chat-style messages\nmanager.dispatch_event(\"event_name\", data)\n\n# Get plugin information\nplugins = manager.list_plugins()\nplugin = manager.get(\"plugin_name\")\n```\n\n### BasePlugin\n\nBase class for all plugins:\n\n```python\nfrom plugflow import BasePlugin\nfrom typing import Optional, Any\n\nclass MyPlugin(BasePlugin):\n    name = \"my_plugin\"           # Required: unique plugin name\n    version = \"1.0.0\"           # Required: plugin version\n    priority = 10               # Optional: loading priority (higher first)\n    dependencies = [\"other\"]    # Optional: plugin dependencies\n    \n    def on_load(self, manager) -> None:\n        \"\"\"Called when plugin is loaded\"\"\"\n        pass\n    \n    def on_unload(self, manager) -> None:\n        \"\"\"Called when plugin is unloaded\"\"\"\n        pass\n    \n    def handle_command(self, command: str, args: str) -> Optional[str]:\n        \"\"\"Handle commands - return result or None if not handled\"\"\"\n        return None\n    \n    def filter_message(self, text: str) -> Optional[str]:\n        \"\"\"Filter/modify messages - return modified text or None\"\"\"\n        return None\n    \n    def on_event(self, event: str, data: Any, manager) -> None:\n        \"\"\"Handle events from other plugins\"\"\"\n        pass\n```\n\n## Advanced Features\n\n### Hot Reload\n\nEnable automatic plugin reloading during development:\n\n```python\nmanager = PluginManager(\n    plugins_paths=[\"plugins/\"],\n    hot_reload=True  # Enable hot reload for development\n)\nmanager.load_all()\n\n# Now edit your plugins - changes will be detected automatically!\n```\n\n### Context Sharing\n\nShare application state and resources with plugins:\n\n```python\n# Create manager with shared context\ncontext = {\n    \"database\": db_connection,\n    \"config\": app_config,\n    \"logger\": logger\n}\nmanager = PluginManager(context=context)\n\n# Plugins can access context\nclass DatabasePlugin(BasePlugin):\n    def on_load(self, manager):\n        db = self.context.get(\"database\")\n        logger = self.context.get(\"logger\")\n        logger.info(\"Database plugin connected!\")\n```\n\n### Event System\n\nPlugins can communicate through events:\n\n```python\n# Plugin A sends event\nclass PublisherPlugin(BasePlugin):\n    def handle_command(self, command, args):\n        if command == \"notify\":\n            # Dispatch event to all plugins\n            self.manager.dispatch_event(\"notification\", {\n                \"message\": args,\n                \"timestamp\": time.time()\n            })\n\n# Plugin B receives event\nclass SubscriberPlugin(BasePlugin):\n    def on_event(self, event, data, manager):\n        if event == \"notification\":\n            print(f\"Received: {data['message']}\")\n```\n\n### Plugin Dependencies\n\nSpecify plugin loading order with dependencies:\n\n```python\nclass DatabasePlugin(BasePlugin):\n    name = \"database\"\n    priority = 100  # Load first\n\nclass UserPlugin(BasePlugin):\n    name = \"user_manager\"\n    dependencies = [\"database\"]  # Load after database\n    \n    def on_load(self, manager):\n        # Database plugin is guaranteed to be loaded\n        db_plugin = manager.get_plugin(\"database\")\n```\n\n### Error Handling\n\nPlugins are isolated - errors don't crash your application:\n\n```python\nclass FaultyPlugin(BasePlugin):\n    def handle_command(self, command, args):\n        if command == \"crash\":\n            raise Exception(\"Plugin error!\")\n        return None\n\n# Manager handles plugin errors gracefully\ntry:\n    result = manager.handle_command(\"crash\", \"\")\nexcept Exception as e:\n    print(f\"Plugin error handled: {e}\")\n# Application continues running\n```\n\n## Example Applications\n\nPlugFlow includes comprehensive examples in the `examples/` directory:\n\n### 1. CLI Tool (`examples/cli_tool/`)\n\nProfessional command-line utility with plugin-based commands:\n\n```bash\ncd examples/cli_tool\npython cli.py help          # Show all commands\npython cli.py hash md5 \"test\"  # Cryptographic operations\npython cli.py tree . 2      # File system utilities\n```\n\n**Features**: Type-safe plugins, clean output, debug mode, comprehensive help\n\n### 2. Telegram Bot (`examples/tg_stub/`)\n\nProduction-ready bot simulation with advanced features:\n\n```bash\ncd examples/tg_stub\npython bot.py              # Clean production mode\npython bot.py --debug      # Development mode with logs\n```\n\n**Features**: Command handling, message filtering, dynamic help, hot reload\n\n### 3. GUI Application (`examples/tk_app/`)\n\nSophisticated tkinter application with plugin integration:\n\n```bash\ncd examples/tk_app\npython app.py\n```\n\n**Features**: Menu integration, real-time logging, file operations, event system\n\n### 4. Web Server (`examples/web_server/`)\n\nFlask-based web application with plugin routes:\n\n```bash\ncd examples/web_server\npython server.py\n```\n\n**Features**: Dynamic routing, middleware, API endpoints, template system\n\n## Plugin Development Guide\n\n### Basic Plugin Structure\n\n```python\nfrom plugflow import BasePlugin\nfrom typing import Optional\n\nclass MyPlugin(BasePlugin):\n    # Plugin metadata\n    name = \"my_plugin\"\n    version = \"1.0.0\"\n    description = \"My awesome plugin\"\n    author = \"Your Name\"\n    \n    def on_load(self, manager):\n        \"\"\"Initialize plugin resources\"\"\"\n        self.data = {}\n        print(f\"{self.name} loaded!\")\n    \n    def on_unload(self, manager):\n        \"\"\"Clean up plugin resources\"\"\"\n        self.data.clear()\n        print(f\"{self.name} unloaded!\")\n    \n    def handle_command(self, command: str, args: str) -> Optional[str]:\n        \"\"\"Process commands\"\"\"\n        commands = {\n            \"status\": self._status,\n            \"set\": self._set_data,\n            \"get\": self._get_data\n        }\n        \n        if command in commands:\n            return commands[command](args)\n        return None\n    \n    def _status(self, args: str) -> str:\n        return f\"Plugin {self.name} v{self.version} - {len(self.data)} items\"\n    \n    def _set_data(self, args: str) -> str:\n        key, value = args.split(\"=\", 1)\n        self.data[key] = value\n        return f\"Set {key} = {value}\"\n    \n    def _get_data(self, args: str) -> str:\n        return self.data.get(args, \"Key not found\")\n```\n\n### Best Practices\n\n1. **Error Handling**: Always handle exceptions gracefully\n2. **Resource Cleanup**: Implement `on_unload` for proper cleanup\n3. **Type Hints**: Use type annotations for better code quality\n4. **Documentation**: Document your plugin's commands and features\n5. **Testing**: Create unit tests for your plugins\n6. **Versioning**: Use semantic versioning for your plugins\n\n### Plugin Testing\n\n```python\nimport unittest\nfrom plugflow import PluginManager\nfrom my_plugin import MyPlugin\n\nclass TestMyPlugin(unittest.TestCase):\n    def setUp(self):\n        self.manager = PluginManager()\n        # Manually add plugin for testing\n        from plugflow.manager import PluginRecord\n        self.plugin = MyPlugin()\n        record = PluginRecord(self.plugin, Path(\"test\"), None)\n        self.manager._records[self.plugin.name] = record\n    \n    def test_command_handling(self):\n        result = self.manager.handle_message(\"/status\")\n        self.assertTrue(any(\"Plugin my_plugin\" in r for r in result))\n    \n    def test_data_operations(self):\n        self.manager.handle_message(\"/set key=value\")\n        result = self.manager.handle_message(\"/get key\")\n        self.assertTrue(any(\"value\" in r for r in result))\n```\n\n## API Reference\n\n### PluginManager\n\n#### Methods\n\n- `load_from_path(path: Path) -> None`: Load plugins from a file or directory\n- `load_all() -> None`: Load all plugins from configured paths\n- `unload_plugin(name: str) -> bool`: Unload a plugin by name\n- `reload_plugin(name: str) -> bool`: Reload a plugin by name\n- `handle_message(text: str) -> List[str]`: Process message through plugins (supports /commands and filters)\n- `dispatch_event(event: str, data: Any = None) -> List[Any]`: Send event to all plugins\n- `broadcast(method: str, *args, **kwargs) -> List[Any]`: Call method on all plugins that have it\n- `list_plugins() -> List[str]`: Get list of loaded plugin names\n- `get(name: str) -> Optional[BasePlugin]`: Get plugin instance by name\n- `stop() -> None`: Stop hot reload watchers\n\n#### Properties\n\n- `hot_reload: bool`: Enable/disable hot reload\n- `context: Dict[str, Any]`: Shared context dictionary\n\n### BasePlugin\n\n#### Required Attributes\n\n- `name: str`: Unique plugin identifier\n- `version: str`: Plugin version\n\n#### Optional Attributes\n\n- `priority: int`: Loading priority (default: 0)\n- `dependencies: List[str]`: Required plugins\n- `description: str`: Plugin description\n- `author: str`: Plugin author\n\n#### Methods\n\n- `on_load(manager: PluginManager) -> None`: Initialization hook\n- `on_unload(manager: PluginManager) -> None`: Cleanup hook\n- `handle_command(command: str, args: str) -> Optional[str]`: Command handler\n- `filter_message(text: str) -> Optional[str]`: Message filter\n- `on_event(event: str, data: Any, manager: PluginManager) -> None`: Event handler\n\n## Performance Tips\n\n1. **Lazy Loading**: Load plugins only when needed\n2. **Caching**: Cache plugin results for expensive operations\n3. **Priority Optimization**: Use priorities to control loading order\n4. **Resource Management**: Properly clean up plugin resources\n5. **Event Filtering**: Only subscribe to relevant events\n\n## Troubleshooting\n\n### Common Issues\n\n**Plugin Not Loading**\n```python\n# Check plugin file syntax\nmanager.load_from_path(Path(\"plugin.py\"))\n\n# Enable debug logging\nimport logging\nlogging.basicConfig(level=logging.DEBUG)\n```\n\n**Import Errors**\n```python\n# Ensure plugin directory is in Python path\nimport sys\nsys.path.append(\"plugins\")\n```\n\n**Hot Reload Not Working**\n```python\n# Ensure hot_reload is enabled\nmanager = PluginManager(hot_reload=True)\n\n# Check file permissions and watching capability\n```\n\n### Debug Mode\n\nEnable verbose logging:\n\n```python\nimport logging\nlogging.basicConfig(level=logging.DEBUG)\n\nmanager = PluginManager(hot_reload=True)\n# Now you'll see detailed plugin loading information\n```\n\n## Contributing\n\nWe welcome contributions!\n\n### Development Setup\n\n```bash\ngit clone https://github.com/keklick1337/plugflow.git\ncd plugflow\npip install -e \".[dev]\"\npre-commit install\n```\n\n### Running Tests\n\n```bash\npytest tests/\n```\n\n## License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n\n---\n\n**Built with love for the Python community**\n\nPlugFlow makes it easy to create extensible applications. Start building your plugin ecosystem today!\n",
    "bugtrack_url": null,
    "license": "MIT",
    "summary": "A powerful Python plugin system with dynamic loading and hot-reload capabilities.",
    "version": "1.0.3",
    "project_urls": {
        "Bug Reports": "https://github.com/keklick1337/plugflow/issues",
        "Documentation": "https://github.com/keklick1337/plugflow",
        "Homepage": "https://github.com/keklick1337/plugflow",
        "Source Code": "https://github.com/keklick1337/plugflow"
    },
    "split_keywords": [
        "plugin",
        " plugins",
        " dynamic-loading",
        " hot-reload",
        " extensible",
        " plugin-system",
        " python-plugins"
    ],
    "urls": [
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "a3340d7891f9afca94d82bb5a3873fdb3bf2df6aab81b59886cd8500c24ba91f",
                "md5": "7261bbfb3e1fb1116508252ec29f580c",
                "sha256": "d44cb821e5ae2a446fa6b3eae6d4089afd360cbf9109f36613da1924f261c044"
            },
            "downloads": -1,
            "filename": "plugflow-1.0.3-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "7261bbfb3e1fb1116508252ec29f580c",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.8",
            "size": 13676,
            "upload_time": "2025-08-26T12:31:48",
            "upload_time_iso_8601": "2025-08-26T12:31:48.673510Z",
            "url": "https://files.pythonhosted.org/packages/a3/34/0d7891f9afca94d82bb5a3873fdb3bf2df6aab81b59886cd8500c24ba91f/plugflow-1.0.3-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "3ef7777322f47868709f18f33dcdf0ddeed2b124d8b83750867dfb96c1671d13",
                "md5": "6bfc7ea2cf407867169c6e43e10e3d52",
                "sha256": "bbc613af473046d744a6428584ad1f74ec528215e52cf2ceee818b8204ac2c02"
            },
            "downloads": -1,
            "filename": "plugflow-1.0.3.tar.gz",
            "has_sig": false,
            "md5_digest": "6bfc7ea2cf407867169c6e43e10e3d52",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.8",
            "size": 26940,
            "upload_time": "2025-08-26T12:31:49",
            "upload_time_iso_8601": "2025-08-26T12:31:49.515558Z",
            "url": "https://files.pythonhosted.org/packages/3e/f7/777322f47868709f18f33dcdf0ddeed2b124d8b83750867dfb96c1671d13/plugflow-1.0.3.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2025-08-26 12:31:49",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "github_user": "keklick1337",
    "github_project": "plugflow",
    "travis_ci": false,
    "coveralls": false,
    "github_actions": true,
    "tox": true,
    "lcname": "plugflow"
}
        
Elapsed time: 1.35584s