adapters

Adapter utilities for making plugin managers compatible with the worker system.

Simple Adapter Factory

---

source

create_simple_adapter

 create_simple_adapter (plugin_manager:Any,
                        result_adapter:Optional[<built-
                        infunctioncallable>]=None)

Create a simple adapter for a plugin manager.

	Type	Default	Details
plugin_manager	Any		The plugin manager instance to adapt
result_adapter	Optional	None	Optional function to convert plugin results to dict
Returns	PluginManagerAdapter		Adapter that satisfies PluginManagerAdapter protocol

The create_simple_adapter function is useful when you have an existing plugin manager that doesn’t explicitly implement the PluginManagerAdapter protocol. It wraps your plugin manager and ensures it has all the required methods.

You can optionally provide a result_adapter function to transform plugin execution results into a standard dictionary format. This is helpful when different plugins return results in different formats.

# Create a mock plugin manager for testing
class MockPluginManager:
    def discover_plugins(self):
        return [type('Plugin', (), {'name': 'test_plugin'})]
    
    def load_plugin(self, plugin_data, config):
        pass
    
    def execute_plugin(self, plugin_name, **params):
        return {'result': f'Executed {plugin_name}'}
    
    def execute_plugin_stream(self, plugin_name, **params):
        yield f'Stream from {plugin_name}'
    
    def reload_plugin(self, plugin_name, config=None):
        pass
    
    def unload_plugin(self, plugin_name):
        pass
    
    def check_streaming_support(self, plugin_name):
        return True

# Test creating an adapter
mock_manager = MockPluginManager()
adapter = create_simple_adapter(mock_manager)
adapter

<__main__.create_simple_adapter.<locals>.SimpleAdapter>

# Test adapter methods
adapter.discover_plugins()

[__main__.Plugin]

# Test execution
adapter.execute_plugin('test_plugin', param1='value1')

{'result': 'Executed test_plugin'}

# Test with custom result adapter
def custom_adapter(result):
    return {'transformed': True, 'original': result}

adapter_with_transform = create_simple_adapter(mock_manager, result_adapter=custom_adapter)
adapter_with_transform.execute_plugin('test_plugin')

{'transformed': True, 'original': {'result': 'Executed test_plugin'}}

Default Result Adapter

---

source

default_result_adapter

 default_result_adapter (result:Any)

Default adapter for converting plugin results to dictionaries.

	Type	Details
result	Any	Plugin execution result
Returns	Dict	Dictionary with text and metadata

The default_result_adapter handles three common cases: 1. Results that are already dictionaries (pass-through) 2. Objects with text and metadata attributes 3. Plain values (converted to string for the text field)

# Test with dict input
result_dict = {'text': 'Hello', 'metadata': {'key': 'value'}}
default_result_adapter(result_dict)

{'text': 'Hello', 'metadata': {'key': 'value'}}

# Test with object that has text and metadata attributes
class DummyResult:
    def __init__(self):
        self.text = "Result text"
        self.metadata = {"source": "test"}

default_result_adapter(DummyResult())

{'text': 'Result text', 'metadata': {'source': 'test'}}

# Test with plain string
default_result_adapter("Plain text result")

{'text': 'Plain text result', 'metadata': {}}