Core API Reference

The pydapter.core module provides the foundational adapter system for converting between Pydantic models and various data formats. It implements a registry-based pattern that enables stateless, bidirectional data transformations.

Installation

pip install pydapter

Overview

The core module establishes the fundamental concepts of pydapter:

• Adapter Protocol: Defines the interface for data conversion
• Registry System: Manages and discovers adapters
• Adaptable Mixin: Provides convenient model integration
• Error Handling: Comprehensive exception hierarchy for debugging

Core Architecture:
┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐
│     Adapter     │  │ AdapterRegistry │  │    Adaptable    │
│   (Protocol)    │  │   (Manager)     │  │    (Mixin)      │
│                 │  │                 │  │                 │
│ from_obj()      │  │ register()      │  │ adapt_from()    │
│ to_obj()        │  │ get()           │  │ adapt_to()      │
│ obj_key         │  │ adapt_from()    │  │                 │
└─────────────────┘  │ adapt_to()      │  └─────────────────┘
                     └─────────────────┘

The system supports both synchronous and asynchronous operations through parallel implementations in pydapter.core and pydapter.async_core.

Core Protocols

Adapter

Module: pydapter.core

Defines the interface for stateless data conversion between Pydantic models and external formats.

Protocol Interface:

@runtime_checkable
class Adapter(Protocol[T]):
    """Stateless conversion helper."""

    obj_key: ClassVar[str]  # Unique identifier for the adapter

    @classmethod
    def from_obj(cls, subj_cls: type[T], obj: Any, /, *, many: bool = False, **kw): ...

    @classmethod
    def to_obj(cls, subj: T | list[T], /, *, many: bool = False, **kw): ...

Key Concepts:

• Stateless: Adapters should not maintain internal state
• Bidirectional: Support both from_obj (import) and to_obj (export)
• Type-safe: Use generic typing for type safety
• Batch Support: Handle single items or collections via many parameter

Implementation Example:

from pydapter.core import Adapter
from pydantic import BaseModel
import json

class JSONAdapter(Adapter):
    obj_key = "json"

    @classmethod
    def from_obj(cls, subj_cls: type[T], obj: str, /, *, many: bool = False, **kw):
        """Convert JSON string to Pydantic model(s)."""
        data = json.loads(obj)
        if many:
            return [subj_cls.model_validate(item) for item in data]
        return subj_cls.model_validate(data)

    @classmethod
    def to_obj(cls, subj: T | list[T], /, *, many: bool = False, **kw):
        """Convert Pydantic model(s) to JSON string."""
        if many or isinstance(subj, list):
            data = [item.model_dump() for item in subj]
        else:
            data = subj.model_dump()
        return json.dumps(data, **kw)

# Usage
class User(BaseModel):
    name: str
    email: str

json_data = '{"name": "John", "email": "john@example.com"}'
user = JSONAdapter.from_obj(User, json_data)
back_to_json = JSONAdapter.to_obj(user)

AsyncAdapter

Module: pydapter.async_core

Asynchronous counterpart to the Adapter protocol for operations requiring async/await.

Protocol Interface:

@runtime_checkable
class AsyncAdapter(Protocol[T]):
    """Stateless, **async** conversion helper."""

    obj_key: ClassVar[str]

    @classmethod
    async def from_obj(
        cls, subj_cls: type[T], obj: Any, /, *, many: bool = False, **kw
    ) -> T | list[T]: ...

    @classmethod
    async def to_obj(cls, subj: T | list[T], /, *, many: bool = False, **kw) -> Any: ...

Implementation Example:

from pydapter.async_core import AsyncAdapter
import aiohttp
import json

class HTTPAPIAdapter(AsyncAdapter):
    obj_key = "http_api"

    @classmethod
    async def from_obj(cls, subj_cls: type[T], obj: str, /, *, many: bool = False, **kw):
        """Fetch data from HTTP API and convert to model(s)."""
        async with aiohttp.ClientSession() as session:
            async with session.get(obj) as response:
                data = await response.json()
                if many:
                    return [subj_cls.model_validate(item) for item in data]
                return subj_cls.model_validate(data)

    @classmethod
    async def to_obj(cls, subj: T | list[T], /, *, many: bool = False, **kw):
        """Convert model(s) to API payload."""
        if many or isinstance(subj, list):
            return [item.model_dump() for item in subj]
        return subj.model_dump()

# Usage
users = await HTTPAPIAdapter.from_obj(User, "https://api.example.com/users", many=True)

Registry System

AdapterRegistry

Module: pydapter.core

Manages adapter registration and provides convenient access methods.

Class Interface:

class AdapterRegistry:
    def __init__(self) -> None: ...

    def register(self, adapter_cls: type[Adapter]) -> None: ...
    def get(self, obj_key: str) -> type[Adapter]: ...
    def adapt_from(self, subj_cls: type[T], obj, *, obj_key: str, **kw): ...
    def adapt_to(self, subj, *, obj_key: str, **kw): ...

Usage:

from pydapter.core import AdapterRegistry

# Create registry
registry = AdapterRegistry()

# Register adapters
registry.register(JSONAdapter)
registry.register(CSVAdapter)

# Use via registry
user = registry.adapt_from(User, json_data, obj_key="json")
csv_data = registry.adapt_to(user, obj_key="csv")

# Direct adapter access
adapter_cls = registry.get("json")
user = adapter_cls.from_obj(User, json_data)

Error Handling:

The registry provides comprehensive error handling:

from pydapter.exceptions import AdapterNotFoundError, AdapterError

try:
    user = registry.adapt_from(User, data, obj_key="unknown")
except AdapterNotFoundError as e:
    print(f"No adapter found: {e}")
except AdapterError as e:
    print(f"Adaptation failed: {e}")

AsyncAdapterRegistry

Module: pydapter.async_core

Asynchronous version of AdapterRegistry for async adapters.

Usage:

from pydapter.async_core import AsyncAdapterRegistry

# Create async registry
async_registry = AsyncAdapterRegistry()
async_registry.register(HTTPAPIAdapter)

# Use with async/await
users = await async_registry.adapt_from(User, api_url, obj_key="http_api", many=True)

Adaptable Mixin

Adaptable

Module: pydapter.core

Mixin class that integrates adapter functionality directly into Pydantic models.

Class Interface:

class Adaptable:
    """Mixin that endows any Pydantic model with adapt-from / adapt-to."""

    _adapter_registry: ClassVar[AdapterRegistry | None] = None

    @classmethod
    def _registry(cls) -> AdapterRegistry: ...

    @classmethod
    def register_adapter(cls, adapter_cls: type[Adapter]) -> None: ...

    @classmethod
    def adapt_from(cls, obj, *, obj_key: str, **kw): ...

    def adapt_to(self, *, obj_key: str, **kw): ...

Usage:

from pydapter.core import Adaptable
from pydantic import BaseModel

class User(BaseModel, Adaptable):
    name: str
    email: str
    age: int

# Register adapters for this model
User.register_adapter(JSONAdapter)
User.register_adapter(CSVAdapter)

# Use adapter methods directly on the model
user = User.adapt_from(json_data, obj_key="json")
csv_output = user.adapt_to(obj_key="csv")

# Class method for creating from external data
users = User.adapt_from(csv_file_content, obj_key="csv", many=True)

Advanced Usage:

# Custom model with multiple adapters
class Product(BaseModel, Adaptable):
    id: str
    name: str
    price: float
    category: str

# Register multiple adapters
Product.register_adapter(JSONAdapter)
Product.register_adapter(XMLAdapter)
Product.register_adapter(DatabaseAdapter)

# Chain conversions
product = Product.adapt_from(xml_data, obj_key="xml")
json_data = product.adapt_to(obj_key="json")
database_record = product.adapt_to(obj_key="database")

Exception Hierarchy

Core Exceptions

Module: pydapter.exceptions

Comprehensive exception system for error handling and debugging.

AdapterError

Base exception for all pydapter errors.

class AdapterError(Exception):
    """Base exception for all pydapter errors."""

    def __init__(self, message: str, **context: Any):
        super().__init__(message)
        self.message = message
        self.context = context  # Additional error context

Usage:

from pydapter.exceptions import AdapterError

try:
    result = adapter.from_obj(Model, invalid_data)
except AdapterError as e:
    print(f"Error: {e.message}")
    print(f"Context: {e.context}")

ValidationError

Exception for data validation failures.

class ValidationError(AdapterError):
    """Exception raised when data validation fails."""

    def __init__(self, message: str, data: Optional[Any] = None, **context: Any):
        super().__init__(message, **context)
        self.data = data  # The data that failed validation

TypeConversionError

Exception for type conversion failures.

class TypeConversionError(ValidationError):
    """Exception raised when type conversion fails."""

    def __init__(
        self,
        message: str,
        source_type: Optional[type] = None,
        target_type: Optional[type] = None,
        field_name: Optional[str] = None,
        model_name: Optional[str] = None,
        **context: Any,
    ): ...

AdapterNotFoundError

Exception when no adapter is registered for a given key.

from pydapter.exceptions import AdapterNotFoundError

try:
    adapter = registry.get("nonexistent")
except AdapterNotFoundError as e:
    print(f"Adapter not found: {e}")

ConfigurationError

Exception for adapter configuration issues.

from pydapter.exceptions import ConfigurationError

class BadAdapter:
    # Missing obj_key will raise ConfigurationError
    pass

try:
    registry.register(BadAdapter)
except ConfigurationError as e:
    print(f"Configuration error: {e}")

Advanced Usage Patterns

Custom Adapter Development

Create specialized adapters for specific use cases:

from pydapter.core import Adapter
from typing import Any, TypeVar
import yaml

T = TypeVar("T")

class YAMLAdapter(Adapter[T]):
    obj_key = "yaml"

    @classmethod
    def from_obj(cls, subj_cls: type[T], obj: str, /, *, many: bool = False, **kw):
        """Convert YAML string to Pydantic model(s)."""
        data = yaml.safe_load(obj)
        if many:
            return [subj_cls.model_validate(item) for item in data]
        return subj_cls.model_validate(data)

    @classmethod
    def to_obj(cls, subj: T | list[T], /, *, many: bool = False, **kw):
        """Convert Pydantic model(s) to YAML string."""
        if many or isinstance(subj, list):
            data = [item.model_dump() for item in subj]
        else:
            data = subj.model_dump()
        return yaml.dump(data, **kw)

Adapter Composition

Combine multiple adapters for complex workflows:

class DataPipeline:
    def __init__(self, model_cls, registry: AdapterRegistry):
        self.model_cls = model_cls
        self.registry = registry

    def transform(self, data, from_format: str, to_format: str, **kw):
        """Transform data from one format to another via Pydantic model."""
        # Parse input format to model
        model_instance = self.registry.adapt_from(
            self.model_cls, data, obj_key=from_format, **kw
        )

        # Convert model to output format
        return self.registry.adapt_to(
            model_instance, obj_key=to_format, **kw
        )

# Usage
pipeline = DataPipeline(User, registry)
json_data = pipeline.transform(csv_data, "csv", "json")

Error Recovery

Implement robust error handling with fallbacks:

def safe_adapt_from(model_cls, data, primary_key: str, fallback_key: str, registry):
    """Attempt adaptation with fallback on failure."""
    try:
        return registry.adapt_from(model_cls, data, obj_key=primary_key)
    except AdapterError as e:
        print(f"Primary adapter {primary_key} failed: {e}")
        try:
            return registry.adapt_from(model_cls, data, obj_key=fallback_key)
        except AdapterError as fallback_error:
            print(f"Fallback adapter {fallback_key} also failed: {fallback_error}")
            raise AdapterError(
                f"Both {primary_key} and {fallback_key} adapters failed",
                primary_error=str(e),
                fallback_error=str(fallback_error)
            )

# Usage
user = safe_adapt_from(User, data, "json", "yaml", registry)

Best Practices

Adapter Design

1. Stateless Design: Keep adapters stateless for thread safety
2. Clear obj_key: Use descriptive, unique keys for adapter identification
3. Error Handling: Provide meaningful error messages with context
4. Type Safety: Use proper type hints and validation
5. Documentation: Document expected input/output formats

Performance Optimization

1. Lazy Loading: Register adapters only when needed
2. Batch Processing: Use many=True for collections
3. Caching: Cache registry lookups for frequently used adapters
4. Memory Management: Be mindful of memory usage with large datasets

Registry Management

1. Global Registry: Use a single global registry for consistency
2. Namespace Keys: Use namespaced keys to avoid conflicts (e.g., "db.postgres")
3. Validation: Validate adapter implementations before registration
4. Testing: Test all registered adapters thoroughly

Error Handling

1. Specific Exceptions: Use specific exception types for different error conditions
2. Context Information: Include relevant context in exception details
3. Logging: Log adapter errors for debugging
4. Recovery Strategies: Implement fallback mechanisms where appropriate

Integration Examples

Database Integration

from pydapter.core import Adapter
import sqlite3

class SQLiteAdapter(Adapter):
    obj_key = "sqlite"

    @classmethod
    def from_obj(cls, subj_cls: type[T], obj: str, /, *, many: bool = False, **kw):
        """Load from SQLite database."""
        conn = sqlite3.connect(kw.get('database', ':memory:'))
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        if many:
            cursor.execute(f"SELECT * FROM {kw.get('table', subj_cls.__name__.lower())}")
            rows = cursor.fetchall()
            return [subj_cls.model_validate(dict(row)) for row in rows]
        else:
            cursor.execute(
                f"SELECT * FROM {kw.get('table', subj_cls.__name__.lower())} WHERE id = ?",
                (kw.get('id'),)
            )
            row = cursor.fetchone()
            return subj_cls.model_validate(dict(row)) if row else None

Web API Integration

from pydapter.async_core import AsyncAdapter
import httpx

class RESTAPIAdapter(AsyncAdapter):
    obj_key = "rest_api"

    @classmethod
    async def from_obj(cls, subj_cls: type[T], obj: str, /, *, many: bool = False, **kw):
        """Fetch from REST API."""
        async with httpx.AsyncClient() as client:
            response = await client.get(obj, params=kw.get('params', {}))
            response.raise_for_status()
            data = response.json()

            if many:
                return [subj_cls.model_validate(item) for item in data]
            return subj_cls.model_validate(data)

    @classmethod
    async def to_obj(cls, subj: T | list[T], /, *, many: bool = False, **kw):
        """Post to REST API."""
        url = kw.get('url')
        if not url:
            raise ValueError("URL required for REST API adapter")

        async with httpx.AsyncClient() as client:
            if many or isinstance(subj, list):
                data = [item.model_dump() for item in subj]
            else:
                data = subj.model_dump()

            response = await client.post(url, json=data)
            response.raise_for_status()
            return response.json()

Migration Guide

When upgrading from previous versions:

1. Adapter Interface: Update custom adapters to use new protocol interface
2. Error Handling: Migrate to new exception hierarchy
3. Registry Usage: Use AdapterRegistry for better organization
4. Async Support: Consider migrating to async adapters for I/O operations
5. Type Safety: Add proper type hints to existing adapters

For detailed migration instructions, see the Migration Guide.

---

Auto-generated API Reference

The following sections contain auto-generated API documentation:

Core Module

pydapter.core

pydapter.core - Adapter protocol, registry, Adaptable mix-in.

Classes

Adaptable

Mixin class that adds adapter functionality to Pydantic models.

This mixin provides convenient methods for converting to/from various data formats by maintaining a registry of adapters and providing high-level convenience methods.

When mixed into a Pydantic model, it adds: - Class methods for registering adapters - Class methods for creating instances from external formats - Instance methods for converting to external formats

Example

from pydantic import BaseModel
from pydapter.core import Adaptable
from pydapter.adapters.json_ import JsonAdapter

class Person(BaseModel, Adaptable):
    name: str
    age: int

# Register an adapter
Person.register_adapter(JsonAdapter)

# Create from JSON
json_data = '{"name": "John", "age": 30}'
person = Person.adapt_from(json_data, obj_key="json")

# Convert to JSON
json_output = person.adapt_to(obj_key="json")

Source code in src/pydapter/core.py

class Adaptable:
    """
    Mixin class that adds adapter functionality to Pydantic models.

    This mixin provides convenient methods for converting to/from various data formats
    by maintaining a registry of adapters and providing high-level convenience methods.

    When mixed into a Pydantic model, it adds:
    - Class methods for registering adapters
    - Class methods for creating instances from external formats
    - Instance methods for converting to external formats

    Example:
        ```python
        from pydantic import BaseModel
        from pydapter.core import Adaptable
        from pydapter.adapters.json_ import JsonAdapter

        class Person(BaseModel, Adaptable):
            name: str
            age: int

        # Register an adapter
        Person.register_adapter(JsonAdapter)

        # Create from JSON
        json_data = '{"name": "John", "age": 30}'
        person = Person.adapt_from(json_data, obj_key="json")

        # Convert to JSON
        json_output = person.adapt_to(obj_key="json")
        ```
    """

    _adapter_registry: ClassVar[AdapterRegistry | None] = None

    @classmethod
    def _registry(cls) -> AdapterRegistry:
        """Get or create the adapter registry for this class."""
        if cls._adapter_registry is None:
            cls._adapter_registry = AdapterRegistry()
        return cls._adapter_registry

    @classmethod
    def register_adapter(cls, adapter_cls: type[Adapter]) -> None:
        """
        Register an adapter class with this model.

        Args:
            adapter_cls: The adapter class to register
        """
        cls._registry().register(adapter_cls)

    @classmethod
    def adapt_from(cls, obj: Any, *, obj_key: str, **kw: Any) -> Any:
        """
        Create model instance(s) from external data format.

        Args:
            obj: The source data in the specified format
            obj_key: The key identifying which adapter to use
            **kw: Additional adapter-specific arguments

        Returns:
            Model instance(s) created from the source data
        """
        return cls._registry().adapt_from(cls, obj, obj_key=obj_key, **kw)

    def adapt_to(self, *, obj_key: str, **kw: Any) -> Any:
        """
        Convert this model instance to external data format.

        Args:
            obj_key: The key identifying which adapter to use
            **kw: Additional adapter-specific arguments

        Returns:
            Data in the specified external format
        """
        return self._registry().adapt_to(self, obj_key=obj_key, **kw)

Functions

adapt_from(obj, *, obj_key, **kw) classmethod

Create model instance(s) from external data format.

Parameters:

Name	Type	Description	Default
obj	Any	The source data in the specified format	required
obj_key	str	The key identifying which adapter to use	required
**kw	Any	Additional adapter-specific arguments	{}

Returns:

Type	Description
Any	Model instance(s) created from the source data

Source code in src/pydapter/core.py

@classmethod
def adapt_from(cls, obj: Any, *, obj_key: str, **kw: Any) -> Any:
    """
    Create model instance(s) from external data format.

    Args:
        obj: The source data in the specified format
        obj_key: The key identifying which adapter to use
        **kw: Additional adapter-specific arguments

    Returns:
        Model instance(s) created from the source data
    """
    return cls._registry().adapt_from(cls, obj, obj_key=obj_key, **kw)

adapt_to(*, obj_key, **kw)

Convert this model instance to external data format.

Parameters:

Name	Type	Description	Default
obj_key	str	The key identifying which adapter to use	required
**kw	Any	Additional adapter-specific arguments	{}

Returns:

Type	Description
Any	Data in the specified external format

Source code in src/pydapter/core.py

def adapt_to(self, *, obj_key: str, **kw: Any) -> Any:
    """
    Convert this model instance to external data format.

    Args:
        obj_key: The key identifying which adapter to use
        **kw: Additional adapter-specific arguments

    Returns:
        Data in the specified external format
    """
    return self._registry().adapt_to(self, obj_key=obj_key, **kw)

register_adapter(adapter_cls) classmethod

Register an adapter class with this model.

Parameters:

Name	Type	Description	Default
adapter_cls	type[Adapter]	The adapter class to register	required

Source code in src/pydapter/core.py

@classmethod
def register_adapter(cls, adapter_cls: type[Adapter]) -> None:
    """
    Register an adapter class with this model.

    Args:
        adapter_cls: The adapter class to register
    """
    cls._registry().register(adapter_cls)

Adapter

Bases: Protocol[T]

Protocol defining the interface for data format adapters.

Adapters are stateless conversion helpers that transform data between Pydantic models and various formats (CSV, JSON, TOML, etc.).

Attributes:

Name	Type	Description
obj_key	str	Unique identifier for the adapter type (e.g., "csv", "json")

Example

from pydantic import BaseModel
from pydapter.adapters.json_ import JsonAdapter

class Person(BaseModel):
    name: str
    age: int

# Convert from JSON to Pydantic model
json_data = '{"name": "John", "age": 30}'
person = JsonAdapter.from_obj(Person, json_data)

# Convert from Pydantic model to JSON
json_output = JsonAdapter.to_obj(person)

Source code in src/pydapter/core.py

@runtime_checkable
class Adapter(Protocol[T]):
    """
    Protocol defining the interface for data format adapters.

    Adapters are stateless conversion helpers that transform data between
    Pydantic models and various formats (CSV, JSON, TOML, etc.).

    Attributes:
        obj_key: Unique identifier for the adapter type (e.g., "csv", "json")

    Example:
        ```python
        from pydantic import BaseModel
        from pydapter.adapters.json_ import JsonAdapter

        class Person(BaseModel):
            name: str
            age: int

        # Convert from JSON to Pydantic model
        json_data = '{"name": "John", "age": 30}'
        person = JsonAdapter.from_obj(Person, json_data)

        # Convert from Pydantic model to JSON
        json_output = JsonAdapter.to_obj(person)
        ```
    """

    obj_key: ClassVar[str]

    @classmethod
    def from_obj(
        cls, subj_cls: type[T], obj: Any, /, *, many: bool = False, **kw: Any
    ) -> T | list[T]:
        """
        Convert from external format to Pydantic model instances.

        Args:
            subj_cls: The Pydantic model class to instantiate
            obj: The source data in the adapter's format
            many: If True, expect/return a list of instances
            **kw: Additional adapter-specific arguments

        Returns:
            Single model instance or list of instances based on 'many' parameter
        """
        ...

    @classmethod
    def to_obj(cls, subj: T | list[T], /, *, many: bool = False, **kw: Any) -> Any:
        """
        Convert from Pydantic model instances to external format.

        Args:
            subj: Single model instance or list of instances
            many: If True, handle as list of instances
            **kw: Additional adapter-specific arguments

        Returns:
            Data in the adapter's external format
        """
        ...

Functions

from_obj(subj_cls, obj, /, *, many=False, **kw) classmethod

Convert from external format to Pydantic model instances.

Parameters:

Name	Type	Description	Default
subj_cls	type[T]	The Pydantic model class to instantiate	required
obj	Any	The source data in the adapter's format	required
many	bool	If True, expect/return a list of instances	False
**kw	Any	Additional adapter-specific arguments	{}

Returns:

Type	Description
T | list[T]	Single model instance or list of instances based on 'many' parameter

Source code in src/pydapter/core.py

@classmethod
def from_obj(
    cls, subj_cls: type[T], obj: Any, /, *, many: bool = False, **kw: Any
) -> T | list[T]:
    """
    Convert from external format to Pydantic model instances.

    Args:
        subj_cls: The Pydantic model class to instantiate
        obj: The source data in the adapter's format
        many: If True, expect/return a list of instances
        **kw: Additional adapter-specific arguments

    Returns:
        Single model instance or list of instances based on 'many' parameter
    """
    ...

to_obj(subj, /, *, many=False, **kw) classmethod

Convert from Pydantic model instances to external format.

Parameters:

Name	Type	Description	Default
subj	T | list[T]	Single model instance or list of instances	required
many	bool	If True, handle as list of instances	False
**kw	Any	Additional adapter-specific arguments	{}

Returns:

Type	Description
Any	Data in the adapter's external format

Source code in src/pydapter/core.py

@classmethod
def to_obj(cls, subj: T | list[T], /, *, many: bool = False, **kw: Any) -> Any:
    """
    Convert from Pydantic model instances to external format.

    Args:
        subj: Single model instance or list of instances
        many: If True, handle as list of instances
        **kw: Additional adapter-specific arguments

    Returns:
        Data in the adapter's external format
    """
    ...

AdapterRegistry

Registry for managing and accessing data format adapters.

The registry maintains a mapping of adapter keys to adapter classes, providing a centralized way to register and retrieve adapters for different data formats.

Example

from pydapter.core import AdapterRegistry
from pydapter.adapters.json_ import JsonAdapter

registry = AdapterRegistry()
registry.register(JsonAdapter)

# Use the registry to adapt data
json_data = '{"name": "John", "age": 30}'
person = registry.adapt_from(Person, json_data, obj_key="json")

Source code in src/pydapter/core.py

class AdapterRegistry:
    """
    Registry for managing and accessing data format adapters.

    The registry maintains a mapping of adapter keys to adapter classes,
    providing a centralized way to register and retrieve adapters for
    different data formats.

    Example:
        ```python
        from pydapter.core import AdapterRegistry
        from pydapter.adapters.json_ import JsonAdapter

        registry = AdapterRegistry()
        registry.register(JsonAdapter)

        # Use the registry to adapt data
        json_data = '{"name": "John", "age": 30}'
        person = registry.adapt_from(Person, json_data, obj_key="json")
        ```
    """

    def __init__(self) -> None:
        """Initialize an empty adapter registry."""
        self._reg: dict[str, type[Adapter]] = {}

    def register(self, adapter_cls: type[Adapter]) -> None:
        """
        Register an adapter class with the registry.

        Args:
            adapter_cls: The adapter class to register. Must have an 'obj_key' attribute.

        Raises:
            ConfigurationError: If the adapter class doesn't define 'obj_key'
        """
        key = getattr(adapter_cls, "obj_key", None)
        if not key:
            raise ConfigurationError(
                "Adapter must define 'obj_key'", adapter_cls=adapter_cls.__name__
            )
        self._reg[key] = adapter_cls

    def get(self, obj_key: str) -> type[Adapter]:
        """
        Retrieve an adapter class by its key.

        Args:
            obj_key: The key identifier for the adapter

        Returns:
            The adapter class associated with the key

        Raises:
            AdapterNotFoundError: If no adapter is registered for the given key
        """
        try:
            return self._reg[obj_key]
        except KeyError as exc:
            raise AdapterNotFoundError(
                f"No adapter registered for '{obj_key}'", obj_key=obj_key
            ) from exc

    def adapt_from(
        self, subj_cls: type[T], obj: Any, *, obj_key: str, **kw: Any
    ) -> T | list[T]:
        """
        Convenience method to convert from external format to Pydantic model.

        Args:
            subj_cls: The Pydantic model class to instantiate
            obj: The source data in the specified format
            obj_key: The key identifying which adapter to use
            **kw: Additional adapter-specific arguments

        Returns:
            Model instance(s) created from the source data

        Raises:
            AdapterNotFoundError: If no adapter is registered for obj_key
            AdapterError: If the adaptation process fails
        """
        try:
            result = self.get(obj_key).from_obj(subj_cls, obj, **kw)
            if result is None:
                raise AdapterError(f"Adapter {obj_key} returned None", adapter=obj_key)
            return result

        except Exception as exc:
            if isinstance(exc, AdapterError) or isinstance(exc, PYDAPTER_PYTHON_ERRORS):
                raise

            raise AdapterError(
                f"Error adapting from {obj_key}", original_error=str(exc)
            ) from exc

    def adapt_to(self, subj: Any, *, obj_key: str, **kw: Any) -> Any:
        """
        Convenience method to convert from Pydantic model to external format.

        Args:
            subj: The model instance(s) to convert
            obj_key: The key identifying which adapter to use
            **kw: Additional adapter-specific arguments

        Returns:
            Data in the specified external format

        Raises:
            AdapterNotFoundError: If no adapter is registered for obj_key
            AdapterError: If the adaptation process fails
        """
        try:
            result = self.get(obj_key).to_obj(subj, **kw)
            if result is None:
                raise AdapterError(f"Adapter {obj_key} returned None", adapter=obj_key)
            return result

        except Exception as exc:
            if isinstance(exc, AdapterError) or isinstance(exc, PYDAPTER_PYTHON_ERRORS):
                raise

            raise AdapterError(
                f"Error adapting to {obj_key}", original_error=str(exc)
            ) from exc

Functions

__init__()

Initialize an empty adapter registry.

Source code in src/pydapter/core.py

def __init__(self) -> None:
    """Initialize an empty adapter registry."""
    self._reg: dict[str, type[Adapter]] = {}

adapt_from(subj_cls, obj, *, obj_key, **kw)

Convenience method to convert from external format to Pydantic model.

Parameters:

Name	Type	Description	Default
subj_cls	type[T]	The Pydantic model class to instantiate	required
obj	Any	The source data in the specified format	required
obj_key	str	The key identifying which adapter to use	required
**kw	Any	Additional adapter-specific arguments	{}

Returns:

Type	Description
T | list[T]	Model instance(s) created from the source data

Raises:

Type	Description
AdapterNotFoundError	If no adapter is registered for obj_key
AdapterError	If the adaptation process fails

Source code in src/pydapter/core.py

def adapt_from(
    self, subj_cls: type[T], obj: Any, *, obj_key: str, **kw: Any
) -> T | list[T]:
    """
    Convenience method to convert from external format to Pydantic model.

    Args:
        subj_cls: The Pydantic model class to instantiate
        obj: The source data in the specified format
        obj_key: The key identifying which adapter to use
        **kw: Additional adapter-specific arguments

    Returns:
        Model instance(s) created from the source data

    Raises:
        AdapterNotFoundError: If no adapter is registered for obj_key
        AdapterError: If the adaptation process fails
    """
    try:
        result = self.get(obj_key).from_obj(subj_cls, obj, **kw)
        if result is None:
            raise AdapterError(f"Adapter {obj_key} returned None", adapter=obj_key)
        return result

    except Exception as exc:
        if isinstance(exc, AdapterError) or isinstance(exc, PYDAPTER_PYTHON_ERRORS):
            raise

        raise AdapterError(
            f"Error adapting from {obj_key}", original_error=str(exc)
        ) from exc

adapt_to(subj, *, obj_key, **kw)

Convenience method to convert from Pydantic model to external format.

Parameters:

Name	Type	Description	Default
subj	Any	The model instance(s) to convert	required
obj_key	str	The key identifying which adapter to use	required
**kw	Any	Additional adapter-specific arguments	{}

Returns:

Type	Description
Any	Data in the specified external format

Raises:

Type	Description
AdapterNotFoundError	If no adapter is registered for obj_key
AdapterError	If the adaptation process fails

Source code in src/pydapter/core.py

def adapt_to(self, subj: Any, *, obj_key: str, **kw: Any) -> Any:
    """
    Convenience method to convert from Pydantic model to external format.

    Args:
        subj: The model instance(s) to convert
        obj_key: The key identifying which adapter to use
        **kw: Additional adapter-specific arguments

    Returns:
        Data in the specified external format

    Raises:
        AdapterNotFoundError: If no adapter is registered for obj_key
        AdapterError: If the adaptation process fails
    """
    try:
        result = self.get(obj_key).to_obj(subj, **kw)
        if result is None:
            raise AdapterError(f"Adapter {obj_key} returned None", adapter=obj_key)
        return result

    except Exception as exc:
        if isinstance(exc, AdapterError) or isinstance(exc, PYDAPTER_PYTHON_ERRORS):
            raise

        raise AdapterError(
            f"Error adapting to {obj_key}", original_error=str(exc)
        ) from exc

get(obj_key)

Retrieve an adapter class by its key.

Parameters:

Name	Type	Description	Default
obj_key	str	The key identifier for the adapter	required

Returns:

Type	Description
type[Adapter]	The adapter class associated with the key

Raises:

Type	Description
AdapterNotFoundError	If no adapter is registered for the given key

Source code in src/pydapter/core.py

def get(self, obj_key: str) -> type[Adapter]:
    """
    Retrieve an adapter class by its key.

    Args:
        obj_key: The key identifier for the adapter

    Returns:
        The adapter class associated with the key

    Raises:
        AdapterNotFoundError: If no adapter is registered for the given key
    """
    try:
        return self._reg[obj_key]
    except KeyError as exc:
        raise AdapterNotFoundError(
            f"No adapter registered for '{obj_key}'", obj_key=obj_key
        ) from exc

register(adapter_cls)

Register an adapter class with the registry.

Parameters:

Name	Type	Description	Default
adapter_cls	type[Adapter]	The adapter class to register. Must have an 'obj_key' attribute.	required

Raises:

Type	Description
ConfigurationError	If the adapter class doesn't define 'obj_key'

Source code in src/pydapter/core.py

def register(self, adapter_cls: type[Adapter]) -> None:
    """
    Register an adapter class with the registry.

    Args:
        adapter_cls: The adapter class to register. Must have an 'obj_key' attribute.

    Raises:
        ConfigurationError: If the adapter class doesn't define 'obj_key'
    """
    key = getattr(adapter_cls, "obj_key", None)
    if not key:
        raise ConfigurationError(
            "Adapter must define 'obj_key'", adapter_cls=adapter_cls.__name__
        )
    self._reg[key] = adapter_cls

Async Core Module

pydapter.async_core

pydapter.async_core - async counterparts to the sync Adapter stack

Classes

AsyncAdaptable

Mixin that endows any Pydantic model with async adapt-from / adapt-to.

Source code in src/pydapter/async_core.py

class AsyncAdaptable:
    """
    Mixin that endows any Pydantic model with async adapt-from / adapt-to.
    """

    _async_registry: ClassVar[AsyncAdapterRegistry | None] = None

    # registry access
    @classmethod
    def _areg(cls) -> AsyncAdapterRegistry:
        if cls._async_registry is None:
            cls._async_registry = AsyncAdapterRegistry()
        return cls._async_registry

    @classmethod
    def register_async_adapter(cls, adapter_cls: type[AsyncAdapter]) -> None:
        cls._areg().register(adapter_cls)

    # helpers
    @classmethod
    async def adapt_from_async(cls, obj, *, obj_key: str, **kw):
        return await cls._areg().adapt_from(cls, obj, obj_key=obj_key, **kw)

    async def adapt_to_async(self, *, obj_key: str, **kw):
        return await self._areg().adapt_to(self, obj_key=obj_key, **kw)

AsyncAdapter

Bases: Protocol[T]

Stateless, async conversion helper.

Source code in src/pydapter/async_core.py

@runtime_checkable
class AsyncAdapter(Protocol[T]):
    """Stateless, **async** conversion helper."""

    obj_key: ClassVar[str]

    @classmethod
    async def from_obj(
        cls, subj_cls: type[T], obj: Any, /, *, many: bool = False, **kw
    ) -> T | list[T]: ...

    @classmethod
    async def to_obj(cls, subj: T | list[T], /, *, many: bool = False, **kw) -> Any: ...

Exceptions

pydapter.exceptions

pydapter.exceptions - Custom exception hierarchy for pydapter.

Classes

AdapterError

Bases: Exception

Base exception for all pydapter errors.

Source code in src/pydapter/exceptions.py

class AdapterError(Exception):
    """Base exception for all pydapter errors."""

    def __init__(self, message: str, **context: Any):
        super().__init__(message)
        self.message = message
        self.context = context

    def __str__(self) -> str:
        context_str = ", ".join(f"{k}={v!r}" for k, v in self.context.items())
        if context_str:
            return f"{self.message} ({context_str})"
        return self.message

AdapterNotFoundError

Bases: AdapterError

Exception raised when an adapter is not found.

Source code in src/pydapter/exceptions.py

class AdapterNotFoundError(AdapterError):
    """Exception raised when an adapter is not found."""

    def __init__(self, message: str, obj_key: Optional[str] = None, **context: Any):
        super().__init__(message, **context)
        self.obj_key = obj_key

ConfigurationError

Bases: AdapterError

Exception raised when adapter configuration is invalid.

Source code in src/pydapter/exceptions.py

class ConfigurationError(AdapterError):
    """Exception raised when adapter configuration is invalid."""

    def __init__(
        self, message: str, config: Optional[dict[str, Any]] = None, **context: Any
    ):
        super().__init__(message, **context)
        self.config = config

ConnectionError

Bases: AdapterError

Exception raised when a connection to a data source fails.

Source code in src/pydapter/exceptions.py

class ConnectionError(AdapterError):
    """Exception raised when a connection to a data source fails."""

    def __init__(
        self,
        message: str,
        adapter: Optional[str] = None,
        url: Optional[str] = None,
        **context: Any,
    ):
        super().__init__(message, **context)
        self.adapter = adapter
        self.url = url

ParseError

Bases: AdapterError

Exception raised when data parsing fails.

Source code in src/pydapter/exceptions.py

class ParseError(AdapterError):
    """Exception raised when data parsing fails."""

    def __init__(self, message: str, source: Optional[str] = None, **context: Any):
        super().__init__(message, **context)
        self.source = source

QueryError

Bases: AdapterError

Exception raised when a query to a data source fails.

Source code in src/pydapter/exceptions.py

class QueryError(AdapterError):
    """Exception raised when a query to a data source fails."""

    def __init__(
        self,
        message: str,
        query: Optional[Any] = None,
        adapter: Optional[str] = None,
        **context: Any,
    ):
        super().__init__(message, **context)
        self.query = query
        self.adapter = adapter

ResourceError

Bases: AdapterError

Exception raised when a resource (file, database, etc.) cannot be accessed.

Source code in src/pydapter/exceptions.py

class ResourceError(AdapterError):
    """Exception raised when a resource (file, database, etc.) cannot be accessed."""

    def __init__(self, message: str, resource: Optional[str] = None, **context: Any):
        super().__init__(message, **context)
        self.resource = resource

TypeConversionError

Bases: ValidationError

Exception raised when type conversion fails.

Source code in src/pydapter/exceptions.py

class TypeConversionError(ValidationError):
    """Exception raised when type conversion fails."""

    def __init__(
        self,
        message: str,
        source_type: Optional[type] = None,
        target_type: Optional[type] = None,
        field_name: Optional[str] = None,
        model_name: Optional[str] = None,
        **context: Any,
    ):
        super().__init__(message, **context)
        self.source_type = source_type
        self.target_type = target_type
        self.field_name = field_name
        self.model_name = model_name

ValidationError

Bases: AdapterError

Exception raised when data validation fails.

Source code in src/pydapter/exceptions.py

class ValidationError(AdapterError):
    """Exception raised when data validation fails."""

    def __init__(self, message: str, data: Optional[Any] = None, **context: Any):
        super().__init__(message, **context)
        self.data = data