Core API Reference

Foundation protocols, exceptions, and plugin system.

Protocols

dspu.core.protocols.Serializable

Bases: Protocol

Protocol for objects that can be serialized to and from dictionaries.

This protocol defines the interface for objects that can be converted to dictionary format for serialization and reconstructed from dictionaries.

Example

class MyData: ... def init(self, value: int): ... self.value = value ... ... def to_dict(self) -> dict[str, Any]: ... return {"value": self.value} ... ... @classmethod ... def from_dict(cls, data: dict[str, Any]) -> Self: ... return cls(data["value"]) ... isinstance(MyData(42), Serializable) True

Functions

to_dict

to_dict() -> dict[str, Any]

Convert object to dictionary representation.

Returns:

Type	Description
dict[str, Any]	Dictionary representation of the object.

Source code in src/dspu/core/protocols.py

def to_dict(self) -> dict[str, Any]:
    """Convert object to dictionary representation.

    Returns:
        Dictionary representation of the object.
    """
    ...

from_dict classmethod

from_dict(data: dict[str, Any]) -> Self

Create object from dictionary representation.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary containing object data.	required

Returns:

Type	Description
Self	New instance of the class.

Source code in src/dspu/core/protocols.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Self:
    """Create object from dictionary representation.

    Args:
        data: Dictionary containing object data.

    Returns:
        New instance of the class.
    """
    ...

dspu.core.protocols.AsyncResource

Bases: Protocol

Protocol for async context managers.

This protocol defines the interface for resources that need async setup and cleanup. It's a stricter version of typing.AsyncContextManager with explicit type hints.

Example

class DatabaseConnection: ... async def aenter(self) -> Self: ... await self.connect() ... return self ... ... async def aexit(self, exc_type, exc_val, exc_tb) -> None: ... await self.disconnect() ... async def use_db(): ... async with DatabaseConnection() as db: ... await db.query("SELECT 1")

dspu.core.protocols.Closeable

Bases: Protocol

Protocol for resources that need explicit cleanup.

This protocol defines the interface for resources that need to be explicitly closed or cleaned up. Supports both sync and async cleanup.

Example

class FileHandle: ... def close(self) -> None: ... # Cleanup logic ... pass ... handle = FileHandle() handle.close() ... class AsyncFileHandle: ... async def aclose(self) -> None: ... # Async cleanup logic ... pass ... async def use_file(): ... handle = AsyncFileHandle() ... await handle.aclose()

Functions

close

close() -> None

Close or cleanup the resource synchronously.

Source code in src/dspu/core/protocols.py

def close(self) -> None:
    """Close or cleanup the resource synchronously."""
    ...

aclose async

aclose() -> None

Close or cleanup the resource asynchronously.

Default implementation calls sync close(). Override for true async cleanup.

Source code in src/dspu/core/protocols.py

async def aclose(self) -> None:
    """Close or cleanup the resource asynchronously.

    Default implementation calls sync close().
    Override for true async cleanup.
    """
    self.close()

Exceptions

dspu.core.exceptions.DSPUError

Bases: Exception

Base exception for all DSPU errors.

All exceptions in the library inherit from this base class, making it easy to catch any DSPU specific error.

Example

try: ... # Some DSPU operation ... pass ... except DSPUError as e: ... print(f"DSPU error: {e}")

dspu.core.exceptions.ConfigurationError

ConfigurationError(
    message: str,
    *,
    field: str | None = None,
    suggestion: str | None = None,
)

Bases: DSPUError

Raised when configuration is invalid or missing.

This exception includes optional field and suggestion information to help users fix configuration issues quickly.

Attributes:

Name	Type	Description
field		The configuration field that caused the error (if applicable).
suggestion		A helpful suggestion for fixing the error.

Example

raise ConfigurationError( ... "Invalid database URL", ... field="database_url", ... suggestion="Use format: postgresql://user:pass@host:port/db" ... )

Initialize configuration error.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the problem.	required
field	str | None	The configuration field that caused the error.	None
suggestion	str | None	A helpful suggestion for fixing the error.	None

Source code in src/dspu/core/exceptions.py

def __init__(
    self,
    message: str,
    *,
    field: str | None = None,
    suggestion: str | None = None,
) -> None:
    """Initialize configuration error.

    Args:
        message: Error message describing the problem.
        field: The configuration field that caused the error.
        suggestion: A helpful suggestion for fixing the error.
    """
    super().__init__(message)
    self.field = field
    self.suggestion = suggestion

dspu.core.exceptions.ValidationError

ValidationError(
    message: str,
    *,
    field: str | None = None,
    value: object = None,
    constraint: str | None = None,
)

Bases: DSPUError

Raised when data validation fails.

This exception is used for data validation errors, including schema validation, type checking, and constraint violations.

Attributes:

Name	Type	Description
field		The field that failed validation (if applicable).
value		The invalid value that caused the error.
constraint		The constraint that was violated.

Example

raise ValidationError( ... "Value must be positive", ... field="age", ... value=-5, ... constraint="age >= 0" ... )

Initialize validation error.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the validation failure.	required
field	str | None	The field that failed validation.	None
value	object	The invalid value.	None
constraint	str | None	The constraint that was violated.	None

Source code in src/dspu/core/exceptions.py

def __init__(
    self,
    message: str,
    *,
    field: str | None = None,
    value: object = None,
    constraint: str | None = None,
) -> None:
    """Initialize validation error.

    Args:
        message: Error message describing the validation failure.
        field: The field that failed validation.
        value: The invalid value.
        constraint: The constraint that was violated.
    """
    super().__init__(message)
    self.field = field
    self.value = value
    self.constraint = constraint

dspu.core.exceptions.DSPUIOError

DSPUIOError(
    message: str,
    *,
    path: str | None = None,
    operation: str | None = None,
)

Bases: DSPUError

Raised when I/O operations fail.

This exception is used for file system, network, and other I/O errors.

Attributes:

Name	Type	Description
path		The path or resource that caused the error.
operation		The operation that failed (e.g., "read", "write").

Example

raise IOError( ... "Failed to read file", ... path="data/file.json", ... operation="read" ... )

Initialize I/O error.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the I/O failure.	required
path	str | None	The path or resource that caused the error.	None
operation	str | None	The operation that failed.	None

Source code in src/dspu/core/exceptions.py

def __init__(
    self,
    message: str,
    *,
    path: str | None = None,
    operation: str | None = None,
) -> None:
    """Initialize I/O error.

    Args:
        message: Error message describing the I/O failure.
        path: The path or resource that caused the error.
        operation: The operation that failed.
    """
    super().__init__(message)
    self.path = path
    self.operation = operation

dspu.core.exceptions.SecurityError

SecurityError(
    message: str,
    *,
    reason: str | None = None,
    resource: str | None = None,
)

Bases: DSPUError

Raised when security operations fail.

This exception is used for authentication, authorization, encryption, and other security-related errors.

Attributes:

Name	Type	Description
reason		The reason for the security failure.
resource		The resource that was being accessed.

Example

raise SecurityError( ... "Authentication failed", ... reason="Invalid credentials", ... resource="database" ... )

Initialize security error.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the security failure.	required
reason	str | None	The reason for the security failure.	None
resource	str | None	The resource that was being accessed.	None

Source code in src/dspu/core/exceptions.py

def __init__(
    self,
    message: str,
    *,
    reason: str | None = None,
    resource: str | None = None,
) -> None:
    """Initialize security error.

    Args:
        message: Error message describing the security failure.
        reason: The reason for the security failure.
        resource: The resource that was being accessed.
    """
    super().__init__(message)
    self.reason = reason
    self.resource = resource

Plugin Registry

dspu.core.registry.Registry

Registry()

Bases: Generic[T]

Generic registry for plugins.

This class provides a simple registry pattern for storing and retrieving plugins by name. It's generic and can store any type.

Example

registry = RegistryCallable registry.register("my_plugin", lambda x: x * 2) func = registry.get("my_plugin") func(5) 10

Initialize empty registry.

Source code in src/dspu/core/registry.py

def __init__(self) -> None:
    """Initialize empty registry."""
    self._plugins: dict[str, T] = {}

Functions

register

register(name: str, plugin: T) -> None

Register a plugin with a name.

Parameters:

Name	Type	Description	Default
name	str	Name to register the plugin under.	required
plugin	T	The plugin instance to register.	required

Raises:

Type	Description
ConfigurationError	If name is already registered.

Example

registry = Registryint registry.register("answer", 42)

Source code in src/dspu/core/registry.py

def register(self, name: str, plugin: T) -> None:
    """Register a plugin with a name.

    Args:
        name: Name to register the plugin under.
        plugin: The plugin instance to register.

    Raises:
        ConfigurationError: If name is already registered.

    Example:
        >>> registry = Registry[int]()
        >>> registry.register("answer", 42)
    """
    if name in self._plugins:
        raise ConfigurationError(
            f"Plugin '{name}' is already registered",
            field="name",
            suggestion=f"Use a different name or unregister '{name}' first",
        )
    self._plugins[name] = plugin

get

get(name: str) -> T

Get a plugin by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the plugin to retrieve.	required

Returns:

Type	Description
T	The registered plugin.

Raises:

Type	Description
ConfigurationError	If name is not registered.

Example

registry = Registryint registry.register("answer", 42) registry.get("answer") 42

Source code in src/dspu/core/registry.py

def get(self, name: str) -> T:
    """Get a plugin by name.

    Args:
        name: Name of the plugin to retrieve.

    Returns:
        The registered plugin.

    Raises:
        ConfigurationError: If name is not registered.

    Example:
        >>> registry = Registry[int]()
        >>> registry.register("answer", 42)
        >>> registry.get("answer")
        42
    """
    if name not in self._plugins:
        available = ", ".join(sorted(self._plugins.keys()))
        suggestion = f"Available plugins: {available}" if available else "No plugins registered"
        raise ConfigurationError(
            f"Plugin '{name}' is not registered",
            field="name",
            suggestion=suggestion,
        )
    return self._plugins[name]

list

list() -> list[str]

List all registered plugin names.

Returns:

Type	Description
list[str]	Sorted list of registered plugin names.

Example

registry = Registryint registry.register("one", 1) registry.register("two", 2) registry.list() ['one', 'two']

Source code in src/dspu/core/registry.py

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

    Returns:
        Sorted list of registered plugin names.

    Example:
        >>> registry = Registry[int]()
        >>> registry.register("one", 1)
        >>> registry.register("two", 2)
        >>> registry.list()
        ['one', 'two']
    """
    return sorted(self._plugins.keys())

Type Definitions

dspu.core.types

Common type definitions for DSPU.

This module defines type aliases and common types used throughout the library.

Usage

from dspu.core import Registry, DSPUError

# Plugin registry
registry = Registry()

@registry.register("my_plugin")
class MyPlugin:
    pass

# Get plugin
PluginClass = registry.get("my_plugin")

# Exception handling
try:
    # Your code
    pass
except DSPUError as e:
    print(f"Error: {e}")

See Also

• Core Examples
• Core User Guide