Index

pyvider.rpcplugin.config

Configuration management for the RPC Plugin framework.

This module provides centralized configuration management using the Foundation framework's configuration system. All defaults are defined in defaults.py following the project's "no inline defaults" policy.

The configuration is organized into separate modules: - runtime.py: Main RPCPluginConfig class with env_field support - configure.py: Configuration helper functions - defaults.py: All default values (no inline defaults) - manager.py: ConfigManager integration for multi-instance management (optional)

Classes

ConfigError

ConfigError(
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any
)

Bases: RPCPluginError

Configuration-related errors in the RPC plugin system.

Raised when there are issues with plugin configuration, including: - Invalid configuration values - Missing required configuration - Type mismatches in configuration - Environment variable parsing errors - Configuration validation failures

Example

from pyvider.rpcplugin.exception import ConfigError

# Raise when required config is missing
if not config.get("plugin_magic_cookie_value"):
    raise ConfigError(
        "Magic cookie value is required",
        hint="Set PLUGIN_MAGIC_COOKIE_VALUE environment variable",
        code="CONFIG_MISSING_COOKIE"
    )

# Raise when config value is invalid
if port < 0 or port > 65535:
    raise ConfigError(
        f"Invalid port number: {port}",
        hint="Port must be between 0 and 65535",
        code="CONFIG_INVALID_PORT"
    )

Source code in pyvider/rpcplugin/exception.py

def __init__(
    self,
    message: str,
    hint: str | None = None,
    code: int | str | None = None,
    *args: Any,
    **kwargs: Any,
) -> None:
    # Store original attributes for backward compatibility
    self.message = message
    self.hint = hint
    # Note: self.code will be set by FoundationError's __init__ to ensure it's always a string

    # Add hint and code to foundation context if provided
    if hint:
        kwargs.setdefault("context", {})["hint"] = hint
    if code is not None:
        kwargs.setdefault("context", {})["error_code"] = code

    # Pass the message and code to FoundationError
    # Convert int codes to strings as required by FoundationError
    super().__init__(message, *args, code=str(code) if code is not None else None, **kwargs)

RPCPluginConfig

Bases: RuntimeConfig

Comprehensive configuration for the RPC plugin system.

This configuration class extends Foundation's RuntimeConfig to provide type-safe, environment-aware configuration for all aspects of the RPC plugin framework. All fields support environment variable overrides with automatic type conversion and validation.

The configuration is organized into functional areas:

Core Settings: - Protocol version negotiation - Magic cookie authentication - Logging configuration

Transport Settings: - Connection and handshake timeouts - Buffer sizes for network I/O - Supported transport mechanisms (Unix, TCP)

Security Settings: - mTLS configuration and auto-generation - Certificate paths and validity periods - Insecure mode for development

gRPC Settings: - Keepalive parameters - Message size limits - Grace periods for shutdown

Client Settings: - Retry logic and backoff strategies - Maximum retry attempts - Transport preferences

Server Settings: - Host and port binding - Unix socket paths - Transport availability

Feature Settings: - Rate limiting with token bucket - Health check service - UI components

Example

from pyvider.rpcplugin.config import rpcplugin_config

# Access configuration values
if rpcplugin_config.plugin_auto_mtls:
    print("mTLS is enabled")

# Modify configuration
rpcplugin_config.plugin_log_level = "DEBUG"
rpcplugin_config.plugin_rate_limit_enabled = True

Note

All configuration fields can be overridden via environment variables. The env_field decorator ensures proper parsing and validation. Default values are defined in the defaults module for consistency.

Functions

clear_plugin_configs

clear_plugin_configs() -> None

Clear all registered RPC plugin configurations.

Warning

This removes all registered configurations. Use with caution.

Example

clear_plugin_configs() assert list_plugin_configs() == []

Source code in pyvider/rpcplugin/config/manager.py

def clear_plugin_configs() -> None:
    """Clear all registered RPC plugin configurations.

    Warning:
        This removes all registered configurations. Use with caution.

    Example:
        >>> clear_plugin_configs()
        >>> assert list_plugin_configs() == []
    """
    manager = get_plugin_config_manager()
    manager.clear()

export_all_plugin_configs

export_all_plugin_configs(
    include_sensitive: bool = False,
) -> dict[str, dict[str, Any]]

Export all registered configurations.

Parameters:

Name	Type	Description	Default
include_sensitive	bool	Whether to include sensitive fields	False

Returns:

Type	Description
dict[str, dict[str, Any]]	Dictionary mapping config names to their dictionaries

Example

all_configs = export_all_plugin_configs() for name, config_dict in all_configs.items(): ... print(f"{name}: {config_dict['plugin_server_port']}")

Source code in pyvider/rpcplugin/config/manager.py

def export_all_plugin_configs(include_sensitive: bool = False) -> dict[str, dict[str, Any]]:
    """Export all registered configurations.

    Args:
        include_sensitive: Whether to include sensitive fields

    Returns:
        Dictionary mapping config names to their dictionaries

    Example:
        >>> all_configs = export_all_plugin_configs()
        >>> for name, config_dict in all_configs.items():
        ...     print(f"{name}: {config_dict['plugin_server_port']}")
    """
    manager = get_plugin_config_manager()
    return manager.export_all(include_sensitive)

export_plugin_config

export_plugin_config(
    name: str, include_sensitive: bool = False
) -> dict[str, Any]

Export a registered configuration as a dictionary.

Parameters:

Name	Type	Description	Default
name	str	Name of the configuration	required
include_sensitive	bool	Whether to include sensitive fields (e.g., passwords)	False

Returns:

Type	Description
dict[str, Any]	Configuration dictionary

Raises:

Type	Description
ValueError	If configuration not found

Example

config_dict = export_plugin_config("server1") import json json.dump(config_dict, open("config.json", "w"))

Source code in pyvider/rpcplugin/config/manager.py

def export_plugin_config(name: str, include_sensitive: bool = False) -> dict[str, Any]:
    """Export a registered configuration as a dictionary.

    Args:
        name: Name of the configuration
        include_sensitive: Whether to include sensitive fields (e.g., passwords)

    Returns:
        Configuration dictionary

    Raises:
        ValueError: If configuration not found

    Example:
        >>> config_dict = export_plugin_config("server1")
        >>> import json
        >>> json.dump(config_dict, open("config.json", "w"))
    """
    manager = get_plugin_config_manager()
    return manager.export(name, include_sensitive)

get_plugin_config

get_plugin_config(name: str) -> RPCPluginConfig | None

Get a registered RPC plugin configuration by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the configuration	required

Returns:

Type	Description
RPCPluginConfig | None	RPCPluginConfig instance or None if not found

Example

config = get_plugin_config("server1") if config: ... print(config.plugin_server_port)

Source code in pyvider/rpcplugin/config/manager.py

def get_plugin_config(name: str) -> RPCPluginConfig | None:
    """Get a registered RPC plugin configuration by name.

    Args:
        name: Name of the configuration

    Returns:
        RPCPluginConfig instance or None if not found

    Example:
        >>> config = get_plugin_config("server1")
        >>> if config:
        ...     print(config.plugin_server_port)
    """
    manager = get_plugin_config_manager()
    return manager.get(name)  # type: ignore[return-value]

get_plugin_config_manager

get_plugin_config_manager() -> ConfigManager

Get the singleton ConfigManager instance for RPC plugins. Returns: ConfigManager instance for RPC plugin configurations

Example

manager = get_plugin_config_manager() manager.register("my-plugin", config=my_config)

Source code in pyvider/rpcplugin/config/manager.py

def get_plugin_config_manager() -> ConfigManager:
    """Get the singleton ConfigManager instance for RPC plugins.
    Returns:
        ConfigManager instance for RPC plugin configurations

    Example:
        >>> manager = get_plugin_config_manager()
        >>> manager.register("my-plugin", config=my_config)
    """
    global _plugin_config_manager
    if _plugin_config_manager is None:
        _plugin_config_manager = ConfigManager()
    return _plugin_config_manager

list_plugin_configs

list_plugin_configs() -> list[str]

List all registered RPC plugin configuration names.

Returns:

Type	Description
list[str]	List of configuration names

Example

configs = list_plugin_configs() for name in configs: ... config = get_plugin_config(name) ... print(f"{name}: port={config.plugin_server_port}")

Source code in pyvider/rpcplugin/config/manager.py

def list_plugin_configs() -> list[str]:
    """List all registered RPC plugin configuration names.

    Returns:
        List of configuration names

    Example:
        >>> configs = list_plugin_configs()
        >>> for name in configs:
        ...     config = get_plugin_config(name)
        ...     print(f"{name}: port={config.plugin_server_port}")
    """
    manager = get_plugin_config_manager()
    return manager.list_configs()

register_plugin_config

register_plugin_config(
    name: str, config: RPCPluginConfig
) -> None

Register an RPC plugin configuration.

Parameters:

Name	Type	Description	Default
name	str	Unique name for this configuration	required
config	RPCPluginConfig	RPCPluginConfig instance to register	required

Raises:

Type	Description
ValueError	If a configuration with this name already exists

Example

from pyvider.rpcplugin.config import RPCPluginConfig config = RPCPluginConfig(plugin_server_port=8080) register_plugin_config("server1", config)

Source code in pyvider/rpcplugin/config/manager.py

def register_plugin_config(
    name: str,
    config: RPCPluginConfig,
) -> None:
    """Register an RPC plugin configuration.

    Args:
        name: Unique name for this configuration
        config: RPCPluginConfig instance to register

    Raises:
        ValueError: If a configuration with this name already exists

    Example:
        >>> from pyvider.rpcplugin.config import RPCPluginConfig
        >>> config = RPCPluginConfig(plugin_server_port=8080)
        >>> register_plugin_config("server1", config)
    """
    manager = get_plugin_config_manager()
    manager.register(name, config=config)

unregister_plugin_config

unregister_plugin_config(name: str) -> None

Unregister an RPC plugin configuration.

Parameters:

Name	Type	Description	Default
name	str	Name of the configuration to remove	required

Example

unregister_plugin_config("server1")

Source code in pyvider/rpcplugin/config/manager.py

def unregister_plugin_config(name: str) -> None:
    """Unregister an RPC plugin configuration.

    Args:
        name: Name of the configuration to remove

    Example:
        >>> unregister_plugin_config("server1")
    """
    manager = get_plugin_config_manager()
    manager.unregister(name)

update_plugin_config

update_plugin_config(
    name: str, updates: dict[str, Any]
) -> None

Update a registered RPC plugin configuration.

Parameters:

Name	Type	Description	Default
name	str	Name of the configuration	required
updates	dict[str, Any]	Dictionary of field updates	required

Raises:

Type	Description
ValueError	If configuration not found

Example

update_plugin_config("server1", {"plugin_server_port": 9000}) config = get_plugin_config("server1") assert config.plugin_server_port == 9000

Source code in pyvider/rpcplugin/config/manager.py

def update_plugin_config(name: str, updates: dict[str, Any]) -> None:
    """Update a registered RPC plugin configuration.

    Args:
        name: Name of the configuration
        updates: Dictionary of field updates

    Raises:
        ValueError: If configuration not found

    Example:
        >>> update_plugin_config("server1", {"plugin_server_port": 9000})
        >>> config = get_plugin_config("server1")
        >>> assert config.plugin_server_port == 9000
    """
    manager = get_plugin_config_manager()
    manager.update(name, updates)