adapters

Common adapters for integrating popular FastHTML libraries with the worker system.

These adapters bridge between common FastHTML libraries and the worker system protocols. They eliminate the need to write custom adapters in each project.

UnifiedPluginRegistry Adapter

Adapter for cjm-fasthtml-plugins UnifiedPluginRegistry to match PluginRegistryProtocol.

---

source

UnifiedPluginRegistryAdapter

 UnifiedPluginRegistryAdapter (registry)

Adapter for cjm-fasthtml-plugins UnifiedPluginRegistry.

	Details
registry	UnifiedPluginRegistry from cjm-fasthtml-plugins

ResourceManager Adapter

Adapter for cjm-fasthtml-resources ResourceManager to match ResourceManagerProtocol.

---

source

ResourceManagerAdapter

 ResourceManagerAdapter (resource_manager)

Adapter for cjm-fasthtml-resources ResourceManager.

	Details
resource_manager	ResourceManager from cjm-fasthtml-resources

SSE Event Broadcaster Adapter

Adapter for cjm-fasthtml-sse SSEBroadcastManager to match EventBroadcasterProtocol.

---

source

SSEBroadcasterAdapter

 SSEBroadcasterAdapter (sse_manager)

Adapter for cjm-fasthtml-sse SSEBroadcastManager.

	Details
sse_manager	SSEBroadcastManager from cjm-fasthtml-sse

Factory Function

Convenience function to create all standard adapters at once.

This factory function creates adapters for: - cjm-fasthtml-plugins UnifiedPluginRegistry - cjm-fasthtml-resources ResourceManager - cjm-fasthtml-sse SSEBroadcastManager

All parameters are optional. Pass None for services you don’t use.

Returns: Tuple of (plugin_adapter, resource_adapter, sse_adapter). Each adapter will be None if the corresponding service was not provided.

Usage Example:

from cjm_fasthtml_workers.extensions.adapters import create_standard_adapters
from cjm_fasthtml_plugins.core.registry import UnifiedPluginRegistry
from cjm_fasthtml_resources.core.manager import ResourceManager

# Create services
plugin_registry = UnifiedPluginRegistry()
resource_manager = ResourceManager()

# Create adapters
plugin_adapter, resource_adapter, sse_adapter = create_standard_adapters(
    plugin_registry=plugin_registry,
    resource_manager=resource_manager
)

# Pass to job manager
manager = MyJobManager(
    plugin_registry=plugin_adapter,
    resource_manager=resource_adapter
)

---

source

create_standard_adapters

 create_standard_adapters (plugin_registry=None, resource_manager=None,
                           sse_manager=None)

Create standard adapters for common FastHTML libraries.

	Type	Default	Details
plugin_registry	NoneType	None	UnifiedPluginRegistry instance (optional)
resource_manager	NoneType	None	ResourceManager instance (optional)
sse_manager	NoneType	None	SSEBroadcastManager instance (optional)
Returns	tuple		(plugin_adapter, resource_adapter, sse_adapter)

Usage Examples

Basic Usage

Create adapters individually:

# Mock registry for testing
class MockPluginRegistry:
    def get_plugins_by_category(self, category):
        return []
    def get_plugin(self, plugin_id):
        return None
    def load_plugin_config(self, plugin_id):
        return {}

# Create adapter
registry = MockPluginRegistry()
adapter = UnifiedPluginRegistryAdapter(registry)

# Use adapter (satisfies PluginRegistryProtocol)
print(adapter.get_plugins_by_category('transcription'))
print(adapter.load_plugin_config('plugin-123'))

[]
{}

Factory Usage

Create all adapters at once:

# Mock services
class MockResourceManager:
    def register_worker(self, pid, worker_type): pass
    def unregister_worker(self, pid): pass
    def update_worker_state(self, pid, **kwargs): pass
    def check_gpu_availability(self): return None
    def get_worker_by_pid(self, pid): return None

class MockSSEManager:
    async def broadcast(self, event_type, data): pass

# Create services
plugin_registry = MockPluginRegistry()
resource_manager = MockResourceManager()
sse_manager = MockSSEManager()

# Create all adapters at once
plugin_adapter, resource_adapter, sse_adapter = create_standard_adapters(
    plugin_registry=plugin_registry,
    resource_manager=resource_manager,
    sse_manager=sse_manager
)

print(f"Plugin adapter: {plugin_adapter}")
print(f"Resource adapter: {resource_adapter}")
print(f"SSE adapter: {sse_adapter}")

Plugin adapter: <__main__.UnifiedPluginRegistryAdapter object>
Resource adapter: <__main__.ResourceManagerAdapter object>
SSE adapter: <__main__.SSEBroadcasterAdapter object>

Partial Usage

Only create adapters for services you use:

# Only using plugin registry and resource manager (no SSE)
plugin_adapter, resource_adapter, sse_adapter = create_standard_adapters(
    plugin_registry=plugin_registry,
    resource_manager=resource_manager
    # sse_manager=None (default)
)

print(f"SSE adapter (should be None): {sse_adapter}")

SSE adapter (should be None): None