Adapters API

This page documents the built-in adapters provided by pydapter.

CSV Adapter

pydapter.adapters.csv_

CSV Adapter for Pydantic Models.

This module provides the CsvAdapter class for converting between Pydantic models and CSV data formats. It supports reading from CSV files or strings and writing Pydantic models to CSV format.

Classes

CsvAdapter

Bases: Adapter[T]

Adapter for converting between Pydantic models and CSV data.

This adapter handles CSV files and strings, providing methods to: - Parse CSV data into Pydantic model instances - Convert Pydantic models to CSV format - Handle various CSV dialects and formatting options

Attributes:

Name	Type	Description
obj_key		The key identifier for this adapter type ("csv")
DEFAULT_CSV_KWARGS		Default CSV parsing parameters

Example

from pydantic import BaseModel
from pydapter.adapters.csv_ import CsvAdapter

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

# Parse CSV data
csv_data = "name,age\nJohn,30\nJane,25"
people = CsvAdapter.from_obj(Person, csv_data, many=True)

# Convert to CSV
csv_output = CsvAdapter.to_obj(people)

Source code in src/pydapter/adapters/csv_.py

class CsvAdapter(Adapter[T]):
    """
    Adapter for converting between Pydantic models and CSV data.

    This adapter handles CSV files and strings, providing methods to:
    - Parse CSV data into Pydantic model instances
    - Convert Pydantic models to CSV format
    - Handle various CSV dialects and formatting options

    Attributes:
        obj_key: The key identifier for this adapter type ("csv")
        DEFAULT_CSV_KWARGS: Default CSV parsing parameters

    Example:
        ```python
        from pydantic import BaseModel
        from pydapter.adapters.csv_ import CsvAdapter

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

        # Parse CSV data
        csv_data = "name,age\\nJohn,30\\nJane,25"
        people = CsvAdapter.from_obj(Person, csv_data, many=True)

        # Convert to CSV
        csv_output = CsvAdapter.to_obj(people)
        ```
    """

    obj_key = "csv"

    # Default CSV dialect settings
    DEFAULT_CSV_KWARGS = {
        "escapechar": "\\",
        "quotechar": '"',
        "delimiter": ",",
        "quoting": csv.QUOTE_MINIMAL,
    }

    # ---------------- incoming
    @classmethod
    def from_obj(
        cls,
        subj_cls: type[T],
        obj: str | Path,
        /,
        *,
        many: bool = True,
        **kw,
    ):
        try:
            # Handle file path or string content
            if isinstance(obj, Path):
                try:
                    text = Path(obj).read_text()
                except Exception as e:
                    raise ParseError(f"Failed to read CSV file: {e}", source=str(obj))
            else:
                text = obj

            # Sanitize text to remove NULL bytes
            text = text.replace("\0", "")

            if not text.strip():
                raise ParseError(
                    "Empty CSV content",
                    source=str(obj)[:100] if isinstance(obj, str) else str(obj),
                )

            # Merge default CSV kwargs with user-provided kwargs
            csv_kwargs = cls.DEFAULT_CSV_KWARGS.copy()
            csv_kwargs.update(kw)  # User-provided kwargs override defaults

            # Parse CSV
            try:
                # Extract specific parameters from csv_kwargs
                delimiter = ","
                quotechar = '"'
                escapechar = "\\"
                quoting = csv.QUOTE_MINIMAL

                if "delimiter" in csv_kwargs:
                    delimiter = str(csv_kwargs.pop("delimiter"))
                if "quotechar" in csv_kwargs:
                    quotechar = str(csv_kwargs.pop("quotechar"))
                if "escapechar" in csv_kwargs:
                    escapechar = str(csv_kwargs.pop("escapechar"))
                if "quoting" in csv_kwargs:
                    quoting_value = csv_kwargs.pop("quoting")
                    if isinstance(quoting_value, int):
                        quoting = quoting_value
                    else:
                        quoting = csv.QUOTE_MINIMAL

                reader = csv.DictReader(
                    io.StringIO(text),
                    delimiter=delimiter,
                    quotechar=quotechar,
                    escapechar=escapechar,
                    quoting=quoting,
                )
                rows = list(reader)

                if not rows:
                    return [] if many else None

                # Check for missing fieldnames
                if not reader.fieldnames:
                    raise ParseError("CSV has no headers", source=text[:100])

                # Check for missing required fields in the model
                model_fields = subj_cls.model_fields
                required_fields = [
                    field for field, info in model_fields.items() if info.is_required()
                ]

                missing_fields = [
                    field for field in required_fields if field not in reader.fieldnames
                ]

                if missing_fields:
                    raise ParseError(
                        f"CSV missing required fields: {', '.join(missing_fields)}",
                        source=text[:100],
                        fields=missing_fields,
                    )

                # Convert rows to model instances
                result = []
                for i, row in enumerate(rows):
                    try:
                        result.append(subj_cls.model_validate(row))
                    except ValidationError as e:
                        raise AdapterValidationError(
                            f"Validation error in row {i + 1}: {e}",
                            data=row,
                            row=i + 1,
                            errors=e.errors(),
                        )

                # If there's only one row and many=False, return a single object
                if len(result) == 1 and not many:
                    return result[0]
                # Otherwise, return a list of objects
                return result

            except csv.Error as e:
                raise ParseError(f"CSV parsing error: {e}", source=text[:100])

        except (ParseError, AdapterValidationError):
            # Re-raise our custom exceptions
            raise
        except Exception as e:
            # Wrap other exceptions
            raise ParseError(
                f"Unexpected error parsing CSV: {e}",
                source=str(obj)[:100] if isinstance(obj, str) else str(obj),
            )

    # ---------------- outgoing
    @classmethod
    def to_obj(
        cls,
        subj: T | list[T],
        /,
        *,
        many: bool = True,
        **kw,
    ) -> str:
        try:
            items = subj if isinstance(subj, list) else [subj]

            if not items:
                return ""

            buf = io.StringIO()

            # Sanitize any string values to remove NULL bytes
            sanitized_items = []
            for item in items:
                item_dict = item.model_dump()
                for key, value in item_dict.items():
                    if isinstance(value, str):
                        item_dict[key] = value.replace("\0", "")
                sanitized_items.append(item_dict)

            # Merge default CSV kwargs with user-provided kwargs
            csv_kwargs = cls.DEFAULT_CSV_KWARGS.copy()
            csv_kwargs.update(kw)  # User-provided kwargs override defaults

            # Get fieldnames from the first item
            fieldnames = list(items[0].model_dump().keys())

            # Extract specific parameters from csv_kwargs
            delimiter = ","
            quotechar = '"'
            escapechar = "\\"
            quoting = csv.QUOTE_MINIMAL

            if "delimiter" in csv_kwargs:
                delimiter = str(csv_kwargs.pop("delimiter"))
            if "quotechar" in csv_kwargs:
                quotechar = str(csv_kwargs.pop("quotechar"))
            if "escapechar" in csv_kwargs:
                escapechar = str(csv_kwargs.pop("escapechar"))
            if "quoting" in csv_kwargs:
                quoting_value = csv_kwargs.pop("quoting")
                if isinstance(quoting_value, int):
                    quoting = quoting_value
                else:
                    quoting = csv.QUOTE_MINIMAL

            writer = csv.DictWriter(
                buf,
                fieldnames=fieldnames,
                delimiter=delimiter,
                quotechar=quotechar,
                escapechar=escapechar,
                quoting=quoting,
            )
            writer.writeheader()
            writer.writerows([i.model_dump() for i in items])
            return buf.getvalue()

        except Exception as e:
            # Wrap exceptions
            raise ParseError(f"Error generating CSV: {e}")

JSON Adapter

pydapter.adapters.json_

JSON Adapter for Pydantic Models.

This module provides the JsonAdapter class for converting between Pydantic models and JSON data formats. It supports reading from JSON files, strings, or bytes and writing Pydantic models to JSON format.

Classes

JsonAdapter

Bases: Adapter[T]

Adapter for converting between Pydantic models and JSON data.

This adapter handles JSON files, strings, and byte data, providing methods to: - Parse JSON data into Pydantic model instances - Convert Pydantic models to JSON format - Handle both single objects and arrays of objects

Attributes:

Name	Type	Description
obj_key		The key identifier for this adapter type ("json")

Example

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

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

# Parse JSON data
json_data = '{"name": "John", "age": 30}'
person = JsonAdapter.from_obj(Person, json_data)

# Parse JSON array
json_array = '[{"name": "John", "age": 30}, {"name": "Jane", "age": 25}]'
people = JsonAdapter.from_obj(Person, json_array, many=True)

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

Source code in src/pydapter/adapters/json_.py

class JsonAdapter(Adapter[T]):
    """
    Adapter for converting between Pydantic models and JSON data.

    This adapter handles JSON files, strings, and byte data, providing methods to:
    - Parse JSON data into Pydantic model instances
    - Convert Pydantic models to JSON format
    - Handle both single objects and arrays of objects

    Attributes:
        obj_key: The key identifier for this adapter type ("json")

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

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

        # Parse JSON data
        json_data = '{"name": "John", "age": 30}'
        person = JsonAdapter.from_obj(Person, json_data)

        # Parse JSON array
        json_array = '[{"name": "John", "age": 30}, {"name": "Jane", "age": 25}]'
        people = JsonAdapter.from_obj(Person, json_array, many=True)

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

    obj_key = "json"

    # ---------------- incoming
    @classmethod
    def from_obj(
        cls, subj_cls: type[T], obj: str | bytes | Path, /, *, many=False, **kw
    ):
        try:
            # Handle file path
            if isinstance(obj, Path):
                try:
                    text = Path(obj).read_text()
                except Exception as e:
                    raise ParseError(f"Failed to read JSON file: {e}", source=str(obj))
            else:
                text = obj.decode("utf-8") if isinstance(obj, bytes) else obj
            # Check for empty input
            if not text or (isinstance(text, str) and not text.strip()):
                raise ParseError(
                    "Empty JSON content",
                    source=str(obj)[:100] if isinstance(obj, str) else str(obj),
                )

            # Parse JSON
            try:
                data = json.loads(text)
            except json.JSONDecodeError as e:
                raise ParseError(
                    f"Invalid JSON: {e}",
                    source=str(text)[:100] if isinstance(text, str) else str(text),
                    position=e.pos,
                    line=e.lineno,
                    column=e.colno,
                )

            # Validate against model
            try:
                if many:
                    if not isinstance(data, list):
                        raise AdapterValidationError(
                            "Expected JSON array for many=True", data=data
                        )
                    return [subj_cls.model_validate(i) for i in data]
                return subj_cls.model_validate(data)
            except ValidationError as e:
                raise AdapterValidationError(
                    f"Validation error: {e}",
                    data=data,
                    errors=e.errors(),
                )

        except (ParseError, AdapterValidationError):
            # Re-raise our custom exceptions
            raise
        except Exception as e:
            # Wrap other exceptions
            raise ParseError(
                f"Unexpected error parsing JSON: {e}",
                source=str(obj)[:100] if isinstance(obj, str) else str(obj),
            )

    # ---------------- outgoing
    @classmethod
    def to_obj(cls, subj: T | list[T], /, *, many=False, **kw) -> str:
        try:
            items = subj if isinstance(subj, list) else [subj]

            if not items:
                return "[]" if many else "{}"

            # Extract JSON serialization options from kwargs
            json_kwargs = {
                "indent": kw.pop("indent", 2),
                "sort_keys": kw.pop("sort_keys", True),
                "ensure_ascii": kw.pop("ensure_ascii", False),
            }

            payload = [i.model_dump() for i in items] if many else items[0].model_dump()
            return json.dumps(payload, **json_kwargs)

        except Exception as e:
            # Wrap exceptions
            raise ParseError(f"Error generating JSON: {e}")

TOML Adapter

pydapter.adapters.toml_

TOML Adapter for Pydantic Models.

This module provides the TomlAdapter class for converting between Pydantic models and TOML data formats. It supports reading from TOML files or strings and writing Pydantic models to TOML format.

Classes

TomlAdapter

Bases: Adapter[T]

Adapter for converting between Pydantic models and TOML data.

This adapter handles TOML files and strings, providing methods to: - Parse TOML data into Pydantic model instances - Convert Pydantic models to TOML format - Handle both single objects and arrays of objects

Attributes:

Name	Type	Description
obj_key		The key identifier for this adapter type ("toml")

Example

from pydantic import BaseModel
from pydapter.adapters.toml_ import TomlAdapter

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

# Parse TOML data
toml_data = '''
name = "John"
age = 30
'''
person = TomlAdapter.from_obj(Person, toml_data)

# Parse TOML array
toml_array = '''
[[people]]
name = "John"
age = 30

[[people]]
name = "Jane"
age = 25
'''
people = TomlAdapter.from_obj(Person, toml_array, many=True)

# Convert to TOML
toml_output = TomlAdapter.to_obj(person)

Source code in src/pydapter/adapters/toml_.py

class TomlAdapter(Adapter[T]):
    """
    Adapter for converting between Pydantic models and TOML data.

    This adapter handles TOML files and strings, providing methods to:
    - Parse TOML data into Pydantic model instances
    - Convert Pydantic models to TOML format
    - Handle both single objects and arrays of objects

    Attributes:
        obj_key: The key identifier for this adapter type ("toml")

    Example:
        ```python
        from pydantic import BaseModel
        from pydapter.adapters.toml_ import TomlAdapter

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

        # Parse TOML data
        toml_data = '''
        name = "John"
        age = 30
        '''
        person = TomlAdapter.from_obj(Person, toml_data)

        # Parse TOML array
        toml_array = '''
        [[people]]
        name = "John"
        age = 30

        [[people]]
        name = "Jane"
        age = 25
        '''
        people = TomlAdapter.from_obj(Person, toml_array, many=True)

        # Convert to TOML
        toml_output = TomlAdapter.to_obj(person)
        ```
    """

    obj_key = "toml"

    @classmethod
    def from_obj(cls, subj_cls: type[T], obj: str | Path, /, *, many=False, **kw):
        try:
            # Handle file path
            if isinstance(obj, Path):
                try:
                    text = Path(obj).read_text()
                except Exception as e:
                    raise ParseError(f"Failed to read TOML file: {e}", source=str(obj))
            else:
                text = obj

            # Check for empty input
            if not text or (isinstance(text, str) and not text.strip()):
                raise ParseError(
                    "Empty TOML content",
                    source=str(obj)[:100] if isinstance(obj, str) else str(obj),
                )

            # Parse TOML
            try:
                parsed = toml.loads(text, **kw)
            except toml.TomlDecodeError as e:
                raise ParseError(
                    f"Invalid TOML: {e}",
                    source=str(text)[:100] if isinstance(text, str) else str(text),
                )

            # Validate against model
            try:
                if many:
                    return [subj_cls.model_validate(x) for x in _ensure_list(parsed)]
                return subj_cls.model_validate(parsed)
            except ValidationError as e:
                raise AdapterValidationError(
                    f"Validation error: {e}",
                    data=parsed,
                    errors=e.errors(),
                )

        except (ParseError, AdapterValidationError):
            # Re-raise our custom exceptions
            raise
        except Exception as e:
            # Wrap other exceptions
            raise ParseError(
                f"Unexpected error parsing TOML: {e}",
                source=str(obj)[:100] if isinstance(obj, str) else str(obj),
            )

    @classmethod
    def to_obj(cls, subj: T | list[T], /, *, many=False, **kw) -> str:
        try:
            items = subj if isinstance(subj, list) else [subj]

            if not items:
                return ""

            payload = (
                {"items": [i.model_dump() for i in items]}
                if many
                else items[0].model_dump()
            )
            return toml.dumps(payload, **kw)

        except Exception as e:
            # Wrap exceptions
            raise ParseError(f"Error generating TOML: {e}")