Skip to content

Core Examples

Examples demonstrating protocols, registry patterns, and exception handling.

Overview

The core module provides foundational utilities including structural typing with protocols, plugin registry, and rich exception handling.

Examples

1. Protocols

File: examples/core/01_protocols_basic.py

Demonstrates structural typing with protocols.

Topics: - Serializable: Objects that can be converted to/from dictionaries - AsyncResource: Objects that support async context management - Closeable: Objects with sync and async cleanup methods

Key Concepts: - Duck typing without inheritance - Runtime protocol checking with isinstance() - Flexible, testable code design

Run: 

uv run python examples/core/01_protocols_basic.py

2. Registry & Plugins

File: examples/core/02_registry_plugins.py

Shows how to build extensible plugin systems.

Topics: - Data transformer plugins - Serializer plugins (JSON, Pickle) - Validator plugins (required fields, types, ranges) - Registry operations (register, get, has, unregister, clear)

Key Concepts: - Plugin architecture without complex frameworks - Type-safe plugin registration - Multiple plugin systems in one application

Run: 

uv run python examples/core/02_registry_plugins.py

3. Exception Handling

File: examples/core/03_exception_handling.py

Demonstrates rich error handling with context.

Topics: - ConfigurationError: Config loading failures with suggestions - ValidationError: Data validation with field, value, and constraint info - DSPUIOError: File operations with path and operation context - SecurityError: Security violations with reason and resource info - Exception hierarchy and catching patterns - Exception chaining

Key Concepts: - Actionable error messages - Rich error context for debugging - Exception chaining to preserve information - Best practices for error handling

Run: 

uv run python examples/core/03_exception_handling.py

Quick Start

from dspu.core import Registry, Serializable
from typing import Protocol

# Define a protocol
class Processor(Protocol):
    def process(self, data: str) -> str: ...

# Create a registry
processors = Registry[Processor]()

# Register plugins
class UpperProcessor:
    def process(self, data: str) -> str:
        return data.upper()

processors.register("upper", UpperProcessor())

# Use plugins
processor = processors.get("upper")
result = processor.process("hello")  # "HELLO"

Common Patterns

Pattern 1: Plugin Architecture

from dspu.core import Registry
from typing import Protocol

# Define plugin interface
class DataTransformer(Protocol):
    def transform(self, data: dict) -> dict: ...

# Create registry
transformers = Registry[DataTransformer]()

# Register transformers
class NormalizeKeys:
    def transform(self, data: dict) -> dict:
        return {k.lower(): v for k, v in data.items()}

class AddTimestamp:
    def transform(self, data: dict) -> dict:
        data["timestamp"] = datetime.now()
        return data

transformers.register("normalize_keys", NormalizeKeys())
transformers.register("add_timestamp", AddTimestamp())

# Use transformers
data = {"Name": "Alice", "AGE": 30}
data = transformers.get("normalize_keys").transform(data)
data = transformers.get("add_timestamp").transform(data)

Pattern 2: Serialization Interface

from dspu.core import Serializable

class User:
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

    def to_dict(self) -> dict:
        return {"name": self.name, "age": self.age}

    @classmethod
    def from_dict(cls, data: dict) -> "User":
        return cls(name=data["name"], age=data["age"])

# User implements Serializable protocol
user = User("Alice", 30)
data = user.to_dict()
restored = User.from_dict(data)

Pattern 3: Rich Error Messages

from dspu.core import ConfigurationError, ValidationError

def load_config(path: str):
    if not path.exists():
        raise ConfigurationError(
            f"Configuration file not found: {path}",
            suggestion="Create the file or check the path"
        )

def validate_user(data: dict):
    if "email" not in data:
        raise ValidationError(
            "Email is required",
            field="email",
            value=None,
            constraint="required"
        )

Pattern 4: Resource Management

from dspu.core import AsyncResource

class Database:
    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.disconnect()

# Database implements AsyncResource protocol
async with Database() as db:
    await db.query("SELECT * FROM users")

Protocol Reference

Serializable

class Serializable(Protocol):
    def to_dict(self) -> dict: ...
    @classmethod
    def from_dict(cls, data: dict) -> Self: ...

AsyncResource

class AsyncResource(Protocol):
    async def __aenter__(self) -> Self: ...
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: ...

Closeable

class Closeable(Protocol):
    def close(self) -> None: ...
    async def aclose(self) -> None: ...

Registry API

# Create registry
registry = Registry[MyProtocol]()

# Register items
registry.register("key", instance)

# Get items
instance = registry.get("key")

# Check existence
if registry.has("key"):
    ...

# List keys
keys = registry.keys()

# Unregister
registry.unregister("key")

# Clear all
registry.clear()

Exception Hierarchy

DSPUError (base)
├── ConfigurationError
├── ValidationError
├── DSPUIOError
│   ├── FormatError
│   └── StorageError
├── SecurityError
└── RetryError

Best Practices

✅ DO: - Use protocols for flexible interfaces - Provide rich error context (field, value, suggestion) - Use registry for plugin systems - Implement protocols implicitly (duck typing) - Chain exceptions to preserve context

❌ DON'T: - Don't use inheritance when protocols suffice - Don't raise generic exceptions without context - Don't forget to document protocol requirements - Don't swallow exceptions without logging

See Also