tessl/pypi-polyfactory

Mock data generation factories for Python types with support for dataclasses, Pydantic, SQLAlchemy, and more

—

Pending

Overview

Eval results

Files

Persistence

Name: tessl/pypi-polyfactory
Author: tessl

Protocols and handlers for persisting generated data to databases and storage systems. Polyfactory provides both synchronous and asynchronous persistence integration, allowing generated instances to be automatically saved during creation.

Capabilities

Synchronous Persistence Protocol

Protocol definition for synchronous persistence handlers that can save single instances or batches of instances.

class SyncPersistenceProtocol:
    """
    Protocol for synchronous persistence operations.
    
    Implement this protocol to create custom persistence handlers
    that can save generated instances to databases or storage systems.
    """
    
    def save(self, data: T) -> T:
        """
        Save a single instance synchronously.
        
        Parameters:
        - data: Instance to persist
        
        Returns:
        The persisted instance (may include database-generated fields)
        """
    
    def save_many(self, data: list[T]) -> list[T]:
        """
        Save multiple instances synchronously.
        
        Parameters:
        - data: List of instances to persist
        
        Returns:
        List of persisted instances
        """

Asynchronous Persistence Protocol

Protocol definition for asynchronous persistence handlers supporting non-blocking database operations.

class AsyncPersistenceProtocol:
    """
    Protocol for asynchronous persistence operations.
    
    Implement this protocol to create custom async persistence handlers
    for non-blocking database operations.
    """
    
    async def save(self, data: T) -> T:
        """
        Save a single instance asynchronously.
        
        Parameters:
        - data: Instance to persist
        
        Returns:
        The persisted instance (may include database-generated fields)
        """
    
    async def save_many(self, data: list[T]) -> list[T]:
        """
        Save multiple instances asynchronously.
        
        Parameters:
        - data: List of instances to persist
        
        Returns:
        List of persisted instances
        """

SQLAlchemy Synchronous Persistence

Built-in persistence handler for SQLAlchemy models with synchronous database operations.

class SQLASyncPersistence:
    """
    Synchronous persistence handler for SQLAlchemy models.
    
    Handles database sessions, commits, and relationship management
    for SQLAlchemy ORM models.
    """
    
    def __init__(self, session: Session) -> None:
        """
        Initialize with SQLAlchemy session.
        
        Parameters:
        - session: SQLAlchemy session for database operations
        """
    
    def save(self, data: T) -> T:
        """Save single SQLAlchemy model instance."""
    
    def save_many(self, data: list[T]) -> list[T]:
        """Save multiple SQLAlchemy model instances."""

Usage Example:

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory, SQLASyncPersistence

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

# Setup database
engine = create_engine('sqlite:///test.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

class UserFactory(SQLAlchemyFactory[User]):
    __model__ = User
    __sync_persistence__ = SQLASyncPersistence(session)

# Create and persist user
user = UserFactory.create_sync()  # Automatically saved to database
users = UserFactory.create_batch_sync(5)  # 5 users saved to database

SQLAlchemy Asynchronous Persistence

Built-in persistence handler for SQLAlchemy models with asynchronous database operations.

class SQLAASyncPersistence:
    """
    Asynchronous persistence handler for SQLAlchemy models.
    
    Handles async database sessions, commits, and relationship management
    for SQLAlchemy ORM models with asyncio support.
    """
    
    def __init__(self, session: AsyncSession) -> None:
        """
        Initialize with SQLAlchemy async session.
        
        Parameters:
        - session: SQLAlchemy async session for database operations
        """
    
    async def save(self, data: T) -> T:
        """Save single SQLAlchemy model instance asynchronously."""
    
    async def save_many(self, data: list[T]) -> list[T]:
        """Save multiple SQLAlchemy model instances asynchronously."""

Usage Example:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory, SQLAASyncPersistence

# Setup async database
engine = create_async_engine('sqlite+aiosqlite:///async_test.db')
AsyncSessionLocal = async_sessionmaker(engine)

class UserFactory(SQLAlchemyFactory[User]):
    __model__ = User
    
    @classmethod
    async def setup_persistence(cls):
        session = AsyncSessionLocal()
        cls.__async_persistence__ = SQLAASyncPersistence(session)

# Usage
await UserFactory.setup_persistence()
user = await UserFactory.create_async()  # Async save to database
users = await UserFactory.create_batch_async(5)  # Async batch save

Custom Persistence Implementation

Create custom persistence handlers by implementing the persistence protocols.

Synchronous Custom Persistence:

from polyfactory import SyncPersistenceProtocol
import json

class JSONFilePersistence(SyncPersistenceProtocol[dict]):
    def __init__(self, file_path: str):
        self.file_path = file_path
        self.data = []
    
    def save(self, data: dict) -> dict:
        # Add ID if not present
        if 'id' not in data:
            data['id'] = len(self.data) + 1
        
        self.data.append(data)
        
        # Save to file
        with open(self.file_path, 'w') as f:
            json.dump(self.data, f, indent=2)
        
        return data
    
    def save_many(self, data: list[dict]) -> list[dict]:
        return [self.save(item) for item in data]

# Usage with TypedDict factory
from typing import TypedDict
from polyfactory.factories import TypedDictFactory

class UserDict(TypedDict):
    name: str
    email: str

class UserDictFactory(TypedDictFactory[UserDict]):
    __model__ = UserDict
    __sync_persistence__ = JSONFilePersistence('users.json')

# Creates user and saves to JSON file
user = UserDictFactory.create_sync()

Asynchronous Custom Persistence:

import asyncio
import aiofiles
import json

class AsyncJSONFilePersistence(AsyncPersistenceProtocol[dict]):
    def __init__(self, file_path: str):
        self.file_path = file_path
        self.data = []
    
    async def save(self, data: dict) -> dict:
        if 'id' not in data:
            data['id'] = len(self.data) + 1
        
        self.data.append(data)
        
        # Async file write
        async with aiofiles.open(self.file_path, 'w') as f:
            await f.write(json.dumps(self.data, indent=2))
        
        return data
    
    async def save_many(self, data: list[dict]) -> list[dict]:
        results = []
        for item in data:
            result = await self.save(item)
            results.append(result)
        return results

class UserDictFactory(TypedDictFactory[UserDict]):
    __model__ = UserDict
    __async_persistence__ = AsyncJSONFilePersistence('async_users.json')

# Async usage
user = await UserDictFactory.create_async()

Factory Persistence Configuration

Configure persistence at the factory class level using class attributes:

class ExampleFactory(BaseFactory[Model]):
    __model__ = Model
    __sync_persistence__: SyncPersistenceProtocol | None = None
    __async_persistence__: AsyncPersistenceProtocol | None = None

Configuration Examples:

# Single persistence handler
class UserFactory(DataclassFactory[User]):
    __model__ = User
    __sync_persistence__ = CustomSyncHandler()

# Both sync and async handlers
class FlexibleFactory(DataclassFactory[User]):
    __model__ = User
    __sync_persistence__ = CustomSyncHandler()
    __async_persistence__ = CustomAsyncHandler()

# Runtime configuration
UserFactory.__sync_persistence__ = new_handler

Error Handling

Persistence operations can raise configuration exceptions when handlers are not properly configured:

from polyfactory.exceptions import ConfigurationException

try:
    # This will raise ConfigurationException if no sync persistence configured
    user = UserFactory.create_sync()
except ConfigurationException as e:
    print(f"Persistence not configured: {e}")

try:
    # This will raise ConfigurationException if no async persistence configured
    user = await UserFactory.create_async()
except ConfigurationException as e:
    print(f"Async persistence not configured: {e}")

Install with Tessl CLI