pydapter 0.3.2

Tiny trait + adapter toolkit for pydantic models

Pydapter: Adapters for Pydantic Models

• docs: https://khive-ai.github.io/pydapter/

Pydapter is a lightweight, type-safe adapter toolkit for Pydantic that enables seamless conversion between Pydantic models and various data formats and storage systems.

Features

• 🔀 Two-way Conversion: Transform data between Pydantic models and external formats/sources
• 🧩 Adaptable Mixin: Add conversion capabilities directly to your models
• 🔄 Async Support: Full support for asynchronous operations
• 📊 Multiple Formats: Support for JSON, CSV, TOML, pandas DataFrames, Excel, and more
• 💾 Database Integration: Connect to PostgreSQL, MongoDB, Neo4j, Qdrant, and others
• 🛡️ Type-Safe: Leverage Pydantic's validation for reliable data handling
• 🚨 Error Handling: Comprehensive error types for different failure scenarios

Installation

# Basic installation
pip install pydapter

# With database support
pip install "pydapter[postgres,mongo,neo4j,qdrant]"

# With pandas support
pip install "pydapter[pandas]"

Quick Start

Basic Adapter Usage

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

# Define a Pydantic model
class User(BaseModel):
    id: int
    name: str
    email: str
    active: bool = True
    tags: List[str] = []

# Create a user
user = User(id=1, name="Alice", email="[e-mail]", tags=["admin"])

# Convert to JSON
json_data = JsonAdapter.to_obj(user)
print(json_data)

# Convert back to model
loaded_user = JsonAdapter.from_obj(User, json_data)
print(loaded_user)

Using the Adaptable Mixin

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

# Define a model with the Adaptable mixin
class Product(BaseModel, Adaptable):
    id: int
    name: str
    price: float
    in_stock: bool = True

# Register adapters
Product.register_adapter(JsonAdapter)
Product.register_adapter(CsvAdapter)

# Create a product
product = Product(id=101, name="Laptop", price=999.99)

# Convert to different formats using the mixin methods
json_data = product.adapt_to(obj_key="json")
csv_data = product.adapt_to(obj_key="csv")

# Convert back to models
product_from_json = Product.adapt_from(json_data, obj_key="json")

Available Adapters

Pydapter includes adapters for various data formats and storage systems:

File Format Adapters

• JsonAdapter: JSON files and strings
• CsvAdapter: CSV files and strings
• TomlAdapter: TOML files and strings

Data Analysis Adapters

• DataFrameAdapter: pandas DataFrame
• SeriesAdapter: pandas Series
• ExcelAdapter: Excel files (requires pandas and openpyxl/xlsxwriter)

Database Adapters

• PostgresAdapter / AsyncPostgresAdapter: PostgreSQL
• MongoAdapter / AsyncMongoAdapter: MongoDB
• Neo4jAdapter: Neo4j graph database
• QdrantAdapter / AsyncQdrantAdapter: Qdrant vector database
• SQLAdapter / AsyncSQLAdapter: Generic SQL (SQLAlchemy)

Detailed Examples

Working with PostgreSQL

from pydantic import BaseModel
from typing import Optional
from pydapter.extras.postgres_ import PostgresAdapter

class User(BaseModel):
    id: Optional[int] = None
    name: str
    email: str

# Read from database
users = PostgresAdapter.from_obj(
    User,
    {
        "engine_url": "postgresql+psycopg://user:pass@localhost/dbname",
        "table": "users",
        "selectors": {"active": True}
    },
    many=True
)

# Store in database
user = User(name="Alice", email="[e-mail]")
PostgresAdapter.to_obj(
    user,
    engine_url="postgresql+psycopg://user:pass@localhost/dbname",
    table="users"
)

Working with MongoDB

from pydantic import BaseModel
from typing import List
from pydapter.extras.mongo_ import MongoAdapter

class Product(BaseModel):
    id: str
    name: str
    price: float
    categories: List[str] = []

# Query from MongoDB
products = MongoAdapter.from_obj(
    Product,
    {
        "url": "mongodb://localhost:27017",
        "db": "shop",
        "collection": "products",
        "filter": {"price": {"$lt": 100}}
    },
    many=True
)

# Store in MongoDB
product = Product(id="prod1", name="Headphones", price=49.99, categories=["audio", "accessories"])
MongoAdapter.to_obj(
    product,
    url="mongodb://localhost:27017",
    db="shop",
    collection="products"
)

Vector Search with Qdrant

from pydantic import BaseModel
from typing import List
from sentence_transformers import SentenceTransformer
from pydapter.extras.qdrant_ import QdrantAdapter

# Load a model to generate embeddings
model = SentenceTransformer('all-MiniLM-L6-v2')

class Document(BaseModel):
    id: str
    title: str
    content: str
    embedding: List[float] = []

    def generate_embedding(self):
        self.embedding = model.encode(self.content).tolist()
        return self

# Create and store a document
doc = Document(
    id="doc1",
    title="Vector Databases",
    content="Vector databases store high-dimensional vectors for similarity search."
).generate_embedding()

QdrantAdapter.to_obj(
    doc,
    collection="documents",
    url="http://localhost:6333"
)

# Search for similar documents
query_vector = model.encode("How do vector databases work?").tolist()
results = QdrantAdapter.from_obj(
    Document,
    {
        "collection": "documents",
        "query_vector": query_vector,
        "top_k": 5,
        "url": "http://localhost:6333"
    },
    many=True
)

Graph Database with Neo4j

from pydantic import BaseModel
from typing import List
from pydapter.extras.neo4j_ import Neo4jAdapter

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

# Store a person in Neo4j
person = Person(id="p1", name="Alice", age=30)
Neo4jAdapter.to_obj(
    person,
    url="bolt://localhost:7687",
    auth=("neo4j", "password"),
    label="Person",
    merge_on="id"
)

# Find people by property
people = Neo4jAdapter.from_obj(
    Person,
    {
        "url": "bolt://localhost:7687",
        "auth": ("neo4j", "password"),
        "label": "Person",
        "where": "n.age > 25"
    },
    many=True
)

Asynchronous Adapters

Many adapters have asynchronous counterparts:

import asyncio
from pydantic import BaseModel
from pydapter.async_core import AsyncAdaptable
from pydapter.extras.async_postgres_ import AsyncPostgresAdapter

class User(BaseModel, AsyncAdaptable):
    id: int
    name: str
    email: str

# Register the async adapter
User.register_async_adapter(AsyncPostgresAdapter)

async def main():
    # Query from database asynchronously
    users = await User.adapt_from_async(
        {
            "engine_url": "postgresql+asyncpg://user:pass@localhost/dbname",
            "table": "users"
        },
        obj_key="async_pg",
        many=True
    )

    # Create a user
    user = User(id=42, name="Bob", email="[e-mail]")

    # Store in database asynchronously
    result = await user.adapt_to_async(
        obj_key="async_pg",
        engine_url="postgresql+asyncpg://user:pass@localhost/dbname",
        table="users"
    )

# Run the async function
asyncio.run(main())

Error Handling

Pydapter provides a rich set of exceptions for detailed error handling:

from pydapter.exceptions import (
    AdapterError, ValidationError, ParseError,
    ConnectionError, QueryError, ResourceError
)
from pydapter.adapters.json_ import JsonAdapter

try:
    # Try to parse invalid JSON
    JsonAdapter.from_obj(User, "{ invalid json }")
except ParseError as e:
    print(f"JSON parsing error: {e}")
except ValidationError as e:
    print(f"Validation error: {e}")
    if hasattr(e, 'errors') and callable(e.errors):
        for error in e.errors():
            print(f"  - {error['loc']}: {error['msg']}")

Extension

Creating your own adapter is straightforward:

from typing import TypeVar
from pydantic import BaseModel
from pydapter.core import Adapter

T = TypeVar("T", bound=BaseModel)

class MyCustomAdapter(Adapter[T]):
    obj_key = "my_format"

    @classmethod
    def from_obj(cls, subj_cls: type[T], obj: Any, /, *, many=False, **kw):
        # Convert from your format to Pydantic models
        ...

    @classmethod
    def to_obj(cls, subj: T | List[T], /, *, many=False, **kw):
        # Convert from Pydantic models to your format
        ...

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Run the CI script locally to ensure all tests pass (python scripts/ci.py)
4. Commit your changes (git commit -m 'Add amazing feature')
5. Push to the branch (git push origin feature/amazing-feature)
6. Open a Pull Request

Continuous Integration

This project uses a comprehensive CI system that runs linting, type checking, unit tests, integration tests, and coverage reporting. The CI script can be run locally to ensure your changes pass all checks before submitting a PR:

# Run all checks
python scripts/ci.py

# Skip integration tests (which require Docker)
python scripts/ci.py --skip-integration

# Run only linting and formatting checks
python scripts/ci.py --skip-unit --skip-integration --skip-type-check --skip-coverage

For more information, see the CI documentation.

License

This project is licensed under the Apache-2 License - see the LICENSE file for details.

Acknowledgements

• Pydantic - The data validation library that makes this possible
• All the amazing database and format libraries this project integrates with

---

Built with ❤️ by the pydapter team