Core Concepts¶

Understanding the foundational patterns and utilities in DSPU.

Overview¶

The core module provides the foundation for DSPU with three key components:

Protocols - Structural typing for flexible interfaces
Registry - Plugin system for extensibility
Exceptions - Rich error handling with context

Protocols¶

What are Protocols?¶

Protocols enable structural (duck) typing in Python. Unlike traditional inheritance, protocols define interfaces based on what methods an object has, not what class it inherits from.

Built-in Protocols¶

Serializable¶

Objects that can be converted to and from dictionaries:

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 automatically satisfies Serializable protocol
# No inheritance needed!

Use Cases: - Configuration objects - Database models - API response objects - Cache serialization

AsyncResource¶

Objects that support async context management:

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

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

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

Use Cases: - Database connections - HTTP clients - File handles - Lock managers

Closeable¶

Objects with cleanup methods:

class Connection:
    def close(self) -> None:
        # Sync cleanup
        ...

    async def aclose(self) -> None:
        # Async cleanup
        ...

# Satisfies Closeable protocol
conn = Connection()
conn.close()  # Or await conn.aclose()

Use Cases: - Resource cleanup - Connection pooling - File handles - Network sockets

Why Protocols?¶

Advantages: - Flexibility: No need for class hierarchies - Testability: Easy to create test doubles - Composition: Mix and match capabilities - Explicit: Clear interface contracts - Type-safe: Static type checking with Pyrefly

Example:

from typing import Protocol

class DataSource(Protocol):
    def fetch(self) -> list[dict]: ...

# Multiple implementations without inheritance
class APISource:
    def fetch(self) -> list[dict]:
        return api_client.get("/data")

class DatabaseSource:
    def fetch(self) -> list[dict]:
        return db.query("SELECT * FROM data")

class FileSource:
    def fetch(self) -> list[dict]:
        return json.load(open("data.json"))

# All satisfy DataSource protocol
def process_data(source: DataSource):
    data = source.fetch()
    # Process data...

Registry¶

What is Registry?¶

Registry provides a type-safe plugin system for building extensible applications without complex frameworks.

Basic Usage¶

from dspu.core import Registry
from typing import Protocol

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

# Create registry
transformers = Registry[Transformer]()

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

class Reverse:
    def transform(self, data: str) -> str:
        return data[::-1]

transformers.register("uppercase", Uppercase())
transformers.register("reverse", Reverse())

# Use plugins
transformer = transformers.get("uppercase")
result = transformer.transform("hello")  # "HELLO"

Registry Operations¶

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

# Get (raises if not found)
instance = registry.get("key")

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

# List keys
keys = registry.keys()

# Unregister
registry.unregister("key")

# Clear all
registry.clear()

Use Cases¶

1. Data Format Registry¶

from typing import Protocol

class Format(Protocol):
    def serialize(self, obj: any) -> str: ...
    def deserialize(self, data: str) -> any: ...

formats = Registry[Format]()
formats.register("json", JSONFormat())
formats.register("yaml", YAMLFormat())
formats.register("toml", TOMLFormat())

# Serialize based on format
fmt = formats.get(format_name)
serialized = fmt.serialize(data)

2. Validator Registry¶

class Validator(Protocol):
    def validate(self, value: any) -> bool: ...

validators = Registry[Validator]()
validators.register("email", EmailValidator())
validators.register("phone", PhoneValidator())
validators.register("url", URLValidator())

# Validate dynamically
for field, value in form_data.items():
    if validators.has(field):
        validator = validators.get(field)
        if not validator.validate(value):
            raise ValidationError(f"Invalid {field}")

3. Command Registry¶

class Command(Protocol):
    def execute(self, args: list[str]) -> None: ...

commands = Registry[Command]()
commands.register("deploy", DeployCommand())
commands.register("test", TestCommand())
commands.register("build", BuildCommand())

# Execute command from CLI
command_name = sys.argv[1]
command = commands.get(command_name)
command.execute(sys.argv[2:])

Best Practices¶

✅ DO: - Use protocols to define plugin interfaces - Keep plugin interfaces small and focused - Document expected plugin behavior - Use meaningful registry keys - Handle missing plugins gracefully

❌ DON'T: - Don't use registry for simple if/else logic - Don't register mutable global state - Don't use string keys without constants - Don't skip error handling for missing plugins

Exception Handling¶

Exception Hierarchy¶

DSPU provides a rich exception hierarchy:

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

Rich Error Context¶

All DSPU exceptions include rich context:

from dspu.core import ConfigurationError, ValidationError

# Configuration error with suggestion
raise ConfigurationError(
    "Database connection failed",
    suggestion="Check DATABASE_URL environment variable"
)

# Validation error with field details
raise ValidationError(
    "Invalid email format",
    field="email",
    value="invalid-email",
    constraint="email_format"
)

Exception Chaining¶

Preserve context when catching and re-raising:

try:
    config = load_config(path)
except FileNotFoundError as e:
    raise ConfigurationError(
        f"Config file not found: {path}",
        suggestion="Create config.yaml or check the path"
    ) from e  # Chain original exception

Best Practices¶

1. Provide Actionable Messages¶

# ❌ Bad: Generic message
raise ConfigurationError("Configuration failed")

# ✅ Good: Specific with suggestion
raise ConfigurationError(
    "Missing required field 'database.host' in config.yaml",
    suggestion="Add 'database.host' to your configuration file"
)

2. Include Context¶

# ❌ Bad: No context
raise ValidationError("Invalid value")

# ✅ Good: Full context
raise ValidationError(
    "Port must be between 1 and 65535",
    field="port",
    value=70000,
    constraint="range(1, 65535)"
)

3. Chain Exceptions¶

# ❌ Bad: Loses original error
try:
    data = json.loads(text)
except json.JSONDecodeError:
    raise FormatError("Invalid JSON")

# ✅ Good: Preserves full traceback
try:
    data = json.loads(text)
except json.JSONDecodeError as e:
    raise FormatError(
        f"Failed to parse JSON from {path}",
        suggestion="Check file format and encoding"
    ) from e

Design Principles¶

DSPU follows these core principles:

1. Explicit Over Implicit¶

Prefer clear, explicit code over "magical" behavior:

# ✅ Good: Explicit
config = Config.load(
    AppConfig,
    sources=[
        FileSource("config.yaml"),
        EnvSource(prefix="APP_"),
    ]
)

# ❌ Bad: Too much magic
config = Config.auto_load()  # Where does config come from?

2. Composition Over Inheritance¶

Use protocols and composition instead of class hierarchies:

# ✅ Good: Composition with protocols
class APIClient:
    def __init__(self, serializer: Serializable):
        self.serializer = serializer

# ❌ Bad: Deep inheritance
class JSONAPIClient(APIClient, JSONMixin, RetryMixin):
    ...

3. Type Safety¶

Leverage Python's type system for safety:

# ✅ Good: Type-safe registry
transformers: Registry[Transformer] = Registry()
transformer = transformers.get("uppercase")  # Type: Transformer

# ❌ Bad: Untyped
transformers = {}  # Type: dict[str, Any]
transformer = transformers["uppercase"]  # Type: Any

4. Fail Fast¶

Validate early and provide clear errors:

# ✅ Good: Validate at load time
config = Config.load(AppConfig, sources=[...])  # Validates immediately

# ❌ Bad: Fail at runtime
config = load_raw_config()  # No validation
port = int(config["port"])  # Fails here with cryptic error

Next Steps¶

Configuration Guide - Multi-source configuration
Core API Reference - Detailed API documentation
Core Examples - Practical examples