API Reference

This page contains the public API reference for pyisolate.

Main Classes

class pyisolate.ExtensionBase

Bases: ExtensionLocal

Base class for all extensions in the pyisolate system.

This is the main class that extension developers should inherit from when creating extensions. It provides the complete extension interface including lifecycle management, RPC communication, and cleanup functionality.

Extensions typically override the lifecycle hooks (before_module_loaded and on_module_loaded) to set up their functionality, and then use the RPC methods to communicate with the host process.

Example

>>> class MyExtension(ExtensionBase):
...     async def on_module_loaded(self, module: ModuleType) -> None:
...         # Set up your extension
...         self.service = module.MyService()
...         self.register_callee(self.service, "my_service")
...
...     async def process_data(self, data: list) -> float:
...         # Extension method callable from host
...         import numpy as np
...         return np.array(data).mean()

_rpc

The AsyncRPC instance for communication (set internally).

__init__() → None

Initialize the extension base class.

This constructor is called automatically when your extension is instantiated. You typically don’t need to override this unless you need to perform initialization before the RPC system is set up.

async stop() → None

Stop the extension and clean up resources.

This method is called by the host when shutting down the extension. It ensures proper cleanup of the RPC communication system and any other resources.

Note

This method is typically called automatically by the ExtensionManager. You should not need to call it directly unless managing extensions manually.

If you need to perform custom cleanup, override before_module_loaded or create a custom cleanup method that is called before stop().

async before_module_loaded() → None

Hook called before the extension module is loaded.

Override this method to perform any setup required before your extension module is imported. This is useful for environment preparation or pre-initialization tasks.

Note

This method is called in the extension’s process, not the host process.

create_caller(object_type: type[proxied_type], object_id: str) → proxied_type

Create a proxy object for calling methods on a remote object.

Use this method to create a caller for objects that exist in the host process. The returned proxy object will forward all async method calls via RPC.

Parameters:

• object_type – The type/interface of the remote object. This is used for type checking and to determine which methods are available.

• object_id – The unique identifier of the remote object to connect to.

Returns:

A proxy object that forwards async method calls to the remote object.

Example

>>> # Create a caller for a service in the host
>>> remote_service = self.create_caller(HostService, "host_service")
>>> result = await remote_service.do_something("data")

async on_module_loaded(module: ModuleType) → None

Hook called after the extension module is successfully loaded.

Override this method to perform initialization that requires access to the loaded module. This is where you typically set up your extension’s main functionality.

Parameters:

module – The loaded Python module object for your extension.

Note

This method is called in the extension’s process, not the host process.

register_callee(object_instance: object, object_id: str) → None

Register an object that can be called remotely from the host process.

Use this method to make your extension’s functionality available to the host process via RPC. The registered object’s async methods can then be called from the host.

Parameters:

• object_instance – The object instance to register for remote calls.

• object_id – A unique identifier for this object. The host will use this ID to create a caller for this object.

Raises:

ValueError – If an object with the given ID is already registered.

Example

>>> class MyService:
...     async def process(self, data: str) -> str:
...         return f"Processed: {data}"
>>>
>>> # In your extension's on_module_loaded:
>>> service = MyService()
>>> self.register_callee(service, "my_service")

use_remote(proxied_singleton: type[ProxiedSingleton]) → None

Configure a ProxiedSingleton class to use remote instances by default.

After calling this method, any instantiation of the singleton class will return a proxy to the remote instance instead of creating a local instance. This is typically used for shared services that should have a single instance across all processes.

Parameters:

proxied_singleton – The ProxiedSingleton class to configure for remote use.

Example

>>> # In your extension's initialization:
>>> self.use_remote(DatabaseSingleton)
>>> # Now DatabaseSingleton() returns a proxy to the host's instance
>>> db = DatabaseSingleton()
>>> await db.set_value("key", "value")

class pyisolate.ExtensionManager(extension_type: type[T], config: ExtensionManagerConfig)

Bases: Generic[T]

Manager for loading and managing extensions in isolated environments.

The ExtensionManager is the primary interface for working with pyisolate. It handles the creation of virtual environments, installation of dependencies, and lifecycle management of extensions. Each extension can run in its own isolated environment with specific dependencies, or share the host environment.

Type Parameters:

T: The base type of extensions this manager will handle. Must be a subclass

of ExtensionBase.

config

The manager configuration containing settings like venv root path.

extensions

Dictionary mapping extension names to their Extension instances.

extension_type

The base extension class type for all managed extensions.

Example

>>> import asyncio
>>> from pyisolate import ExtensionManager, ExtensionManagerConfig, ExtensionConfig
>>>
>>> async def main():
...     # Create manager configuration
...     manager_config = ExtensionManagerConfig(
...         venv_root_path="./my-extensions"
...     )
...
...     # Create manager for a specific extension type
...     manager = ExtensionManager(MyExtensionBase, manager_config)
...
...     # Load an extension
...     ext_config = ExtensionConfig(
...         name="processor",
...         module_path="./extensions/processor",
...         isolated=True,
...         dependencies=["numpy>=1.26.0"],
...         apis=[],
...         share_torch=False
...     )
...     extension = manager.load_extension(ext_config)
...
...     # Use the extension
...     result = await extension.process([1, 2, 3, 4, 5])
...     print(result)
...
...     # Clean up
...     await extension.stop()
>>>
>>> asyncio.run(main())

__init__(extension_type: type[T], config: ExtensionManagerConfig) → None

Initialize the ExtensionManager.

Parameters:

• extension_type – The base class that all extensions managed by this manager should inherit from. This is used for type checking and to ensure extensions have the correct interface.

• config – Configuration for the manager, including the root path for virtual environments.

Raises:

ValueError – If the venv_root_path in config is invalid or not writable.

load_extension(config: ExtensionConfig) → T

Load an extension with the specified configuration.

This method creates a new extension instance, sets up its virtual environment (if isolated), installs dependencies, and establishes RPC communication. The returned object is a proxy that forwards method calls to the extension running in its separate process.

Parameters:

config – Configuration for the extension, including name, module path, dependencies, and isolation settings.

Returns:

A proxy object that implements the extension interface. All async method calls on this object are forwarded to the actual extension via RPC.

Raises:

• ValueError – If an extension with the same name is already loaded, or if the extension name or dependencies contain invalid characters.

• FileNotFoundError – If the module_path doesn’t exist.

• subprocess.CalledProcessError – If dependency installation fails.

• ImportError – If the extension module cannot be imported.

Example

>>> config = ExtensionConfig(
...     name="data_processor",
...     module_path="./extensions/processor",
...     isolated=True,
...     dependencies=["pandas>=2.0.0"],
...     apis=[DatabaseAPI],
...     share_torch=False
... )
>>> extension = manager.load_extension(config)
>>> # Now you can call methods on the extension
>>> result = await extension.process_data(my_data)

Note

The extension process starts immediately upon loading. To stop the extension and clean up resources, call the stop() method on the returned proxy object.

stop_extension(name: str) → None

Stop a specific extension by name.

Parameters:

name – The name of the extension to stop (as provided in ExtensionConfig).

Raises:

KeyError – If no extension with the given name is loaded.

stop_all_extensions() → None

Stop all loaded extensions and clean up resources.

This method stops all extension processes that were loaded by this manager, cleaning up their virtual environments and RPC connections. It’s recommended to call this method before shutting down the application to ensure clean termination of all extension processes.

Configuration

class pyisolate.ExtensionManagerConfig

Bases: TypedDict

Configuration for the ExtensionManager.

This configuration controls the behavior of the ExtensionManager, which is responsible for creating and managing multiple extensions.

Example

>>> config = ExtensionManagerConfig(
...     venv_root_path="/path/to/extension-venvs"
... )
>>> manager = ExtensionManager(MyExtensionBase, config)

venv_root_path: str

The root directory where virtual environments for isolated extensions will be created.

Each extension gets its own subdirectory under this path. The path should be writable and have sufficient space for installing dependencies.

class pyisolate.ExtensionConfig

Bases: TypedDict

Configuration for a specific extension.

This configuration defines how an individual extension should be loaded and managed by the ExtensionManager. It controls isolation, dependencies, and shared resources.

Example

>>> config = ExtensionConfig(
...     name="data_processor",
...     module_path="./extensions/processor",
...     isolated=True,
...     dependencies=["numpy>=1.26.0", "pandas>=2.0.0"],
...     apis=[DatabaseAPI, ConfigAPI],
...     share_torch=False
... )
>>> extension = manager.load_extension(config)

name: str

A unique name for this extension.

This will be used as the directory name for the virtual environment (after normalization for filesystem safety). Should be descriptive and unique within your application.

module_path: str

The filesystem path to the extension package directory.

This must be a directory containing an __init__.py file. The path can be absolute or relative to the current working directory.

isolated: bool

Whether to run this extension in an isolated virtual environment.

If True, a separate venv is created with the specified dependencies. If False, the extension runs in the host Python environment.

dependencies: list[str]

List of pip-installable dependencies for this extension.

Each string should be a valid pip requirement specifier (e.g., “numpy>=1.21.0”, “requests~=2.28.0”). Dependencies are installed in the order specified.

Security Note: The ExtensionManager validates dependencies to prevent command injection, but you should still review dependency lists from untrusted sources.

apis: list[type[ProxiedSingleton]]

List of ProxiedSingleton classes that this extension should have access to.

These singletons will be automatically configured to use remote instances from the host process, enabling shared state across all extensions.

share_torch: bool

Whether to share PyTorch with the host process.

If True, the extension will use torch.multiprocessing for process creation and the exact same PyTorch version as the host. This enables zero-copy tensor sharing between processes. If False, the extension can install its own PyTorch version if needed.

Utilities

class pyisolate.ProxiedSingleton(*args, **kwargs)

Bases: object

Base class for creating shared singleton services across processes.

ProxiedSingleton enables you to create services that have a single instance shared across all extensions and the host process. When an extension accesses a ProxiedSingleton, it automatically gets a proxy to the singleton instance in the host process, ensuring all processes share the same state.

This is particularly useful for shared resources like databases, configuration managers, or any service that should maintain consistent state across all extensions.

Advanced usage: Methods can be marked to run locally in each process instead of being proxied to the host (see internal documentation for details).

Example

>>> from pyisolate import ProxiedSingleton
>>>
>>> class DatabaseService(ProxiedSingleton):
...     def __init__(self):
...         super().__init__()
...         self.data = {}
...
...     async def get(self, key: str) -> Any:
...         return self.data.get(key)
...
...     async def set(self, key: str, value: Any) -> None:
...         self.data[key] = value
...
>>>
>>> # In extension configuration:
>>> config = ExtensionConfig(
...     name="my_extension",
...     module_path="./extension.py",
...     apis=[DatabaseService],  # Grant access to this singleton
...     # ... other config
... )

Note

All methods that should be accessible via RPC must be async methods. Synchronous methods can only be used if marked with @local_execution.

__init__()

Initialize the ProxiedSingleton.

This constructor is called only once per singleton class in the host process. Extensions will receive a proxy instead of creating new instances.

classmethod get_remote_id() → str

Get the unique identifier for this singleton in the RPC system.

By default, this returns the class name. Override this method if you need a different identifier (e.g., to avoid naming conflicts).

You probably don’t need to override this.

Returns:

The string identifier used to register and look up this singleton in the RPC system.