pydantic-ddd-interface 0.1.0

A Python library providing interface definitions for Domain-Driven Design (DDD) patterns using Pydantic for type safety and validation.

🏛️ Pydddi

Pydantic-powered Domain-Driven Design Interfaces

A modern Python library for building clean, type-safe DDD applications

---

Transform your Python applications with enterprise-grade DDD patterns 🚀

✨ What is Pydddi?

Pydddi brings the power of Domain-Driven Design to Python with type safety at its core. Built on top of Pydantic, it provides clean interfaces and patterns that make your code more maintainable, testable, and scalable.

🎯 Why Choose Pydddi?

Feature	Benefit
🔒 Type Safety	Catch errors at development time with Pydantic validation
🏗️ Clean Architecture	Enforce separation of concerns with clear layer boundaries
🧩 Modular Design	Mix and match interfaces based on your needs
⚡ Modern Python	Leverages Python 3.12+ features for optimal performance
📖 Self-Documenting	Interfaces serve as living documentation

🚀 Features

🏛️ Domain Layer Type-safe entities with identity Value objects using Pydantic Domain service abstractions Rich domain modeling	⚙️ Application Layer Command/Result pattern Use case interfaces Structured error handling Business workflow orchestration
🗄️ Infrastructure Layer Repository patterns (CRUD, Read, Aggregate) Schema-based operations Database abstraction External service integration	🧪 Developer Experience Full type hints support IDE autocompletion Runtime validation Clear error messages

📦 Quick Start

# Install the package
pip install pydantic-ddd-interface

# Start building your domain
from pydddi import IEntity, IUseCase, ICrudRepository

# You're ready to build enterprise-grade applications! 🎉

---

🔧 Usage Examples

💡 Tip: Each layer has a specific responsibility. Start with the domain and work your way out!

🏛️ Domain Layer

Define your business entities and value objects

from pydddi import IEntity, IModel

class UserId(IModel):
    """Strongly-typed user identifier"""
    value: int

class User(IEntity):
    """User aggregate root"""
    id: UserId
    name: str
    email: str
    
    def get_id(self) -> UserId:
        return self.id

⚙️ Application Layer

Orchestrate business workflows with use cases

from pydddi import IUseCase, IUseCaseCommand, IUseCaseResult

class CreateUserCommand(IUseCaseCommand):
    """Command to create a new user"""
    name: str
    email: str

class CreateUserResult(IUseCaseResult):
    """Result of user creation"""
    user_id: UserId
    success: bool
    message: str

class CreateUserUseCase(IUseCase[CreateUserCommand, CreateUserResult]):
    """Use case for creating users with business validation"""
    
    async def execute(self, command: CreateUserCommand) -> CreateUserResult:
        # Your business logic here
        # - Validate email uniqueness
        # - Apply business rules
        # - Persist user
        pass

🗄️ Infrastructure Layer

Handle data persistence and external integrations

from pydddi import ICrudRepository, IReadRepository, IReadAggregateRepository
from pydddi import ICreateSchema, IUpdateSchema

class UserCreateSchema(ICreateSchema):
    """Schema for user creation"""
    name: str
    email: str

class UserUpdateSchema(IUpdateSchema):
    """Schema for user updates"""
    name: str | None = None
    email: str | None = None

🔍 Click to see repository implementations

# 🔨 CRUD Repository - for full CRUD operations on entities
class UserCrudRepository(ICrudRepository[User, UserCreateSchema, User, UserUpdateSchema]):
    async def create(self, schema: UserCreateSchema) -> User:
        """Create a new user entity"""
        pass
    
    async def read(self, id: int) -> User:
        """Get user entity by ID (no relationships)"""
        pass
    
    # ... other CRUD methods

# 📖 Read Repository - for simple read operations on entities  
class UserReadRepository(IReadRepository[User, User]):
    async def read(self, id: int) -> User:
        """Get user entity without relationships"""
        pass
    
    async def list(self, limit: int = None, **filters) -> List[User]:
        """List user entities (lightweight)"""
        pass

# 🔗 Read Aggregate Repository - for complex reads with relationships
class UserAggregateRepository(IReadAggregateRepository[UserWithPosts, User]):
    async def read(self, id: int) -> UserWithPosts:
        """Get user with all related data (posts, comments, etc.)"""
        pass
    
    async def list(self, limit: int = None, **filters) -> List[UserWithPosts]:
        """List users with full relationship data"""
        pass

---

🏗️ Architecture Overview

graph TB
    subgraph "🖥️ Presentation Layer"
        API[REST API / GraphQL]
        CLI[CLI Commands]
    end
    
    subgraph "⚙️ Application Layer"
        UC[Use Cases]
        CMD[Commands & Results]
    end
    
    subgraph "🏛️ Domain Layer"
        ENT[Entities]
        VO[Value Objects]
        DS[Domain Services]
    end
    
    subgraph "🗄️ Infrastructure Layer"
        REPO[Repositories]
        DB[(Database)]
        EXT[External APIs]
    end
    
    API --> UC
    CLI --> UC
    UC --> ENT
    UC --> REPO
    REPO --> DB
    REPO --> EXT
    
    style UC fill:#e1f5fe
    style ENT fill:#f3e5f5
    style REPO fill:#e8f5e8

🎯 Design Principles

Principle	Implementation
🔒 Dependency Inversion	Infrastructure depends on domain abstractions
🧩 Single Responsibility	Each interface has one clear purpose
⚡ Open/Closed	Extend behavior without modifying existing code
🎭 Interface Segregation	Small, focused interfaces over large ones

---

📚 API Reference

🏛️ Domain Layer APIs

Entities & Models

• IEntity - Base class for entities with identity and lifecycle
• IModel - Base class for value objects using Pydantic validation
• IDomainService - Interface for domain services containing business logic

⚙️ Application Layer APIs

Use Cases & Commands

• IUseCase[TCommand, TResult] - Generic interface for business use cases
• IUseCaseCommand - Base class for use case input commands
• IUseCaseResult - Base class for use case output results

Exception Handling

• UseCaseError - Base exception for use case operations
• UseCaseCommandError - Command validation errors
• UseCaseResultError - Result validation errors
• UseCaseExecutionError - Execution-time errors

🗄️ Infrastructure Layer APIs

Repository Patterns

• ICrudRepository[TEntity, TCreateSchema, TReadSchema, TUpdateSchema] - Full CRUD operations
• IReadRepository[TEntity, TReadSchema] - Read-only operations for entities
• IReadAggregateRepository[TModel, TReadAggregateSchema] - Complex reads with relationships

Data Schemas

• ICreateSchema - Schema for create operations
• IReadSchema - Schema for read operations
• IUpdateSchema - Schema for update operations
• IReadAggregateSchema - Schema for aggregate read operations

Exception Handling

• RepositoryError - Base repository exception
• RecordNotFoundError - Entity not found errors
• DuplicateRecordError - Duplicate entity errors

🎯 Repository Usage Guide

Repository Type	Use Case	Returns	Example
ICrudRepository	Full entity lifecycle	TEntity	User CRUD operations
IReadRepository	Simple entity queries	TEntity	User profile lookup
IReadAggregateRepository	Complex queries with joins	TModel	User dashboard with posts

---

🤝 Contributing

We welcome contributions! Here's how you can help:

🐛 Found a Bug?	💡 Have an Idea?	📖 Improve Docs?
Open an Issue	Start a Discussion	Submit a PR

🛠️ Development Setup

# Clone the repository
git clone https://github.com/nodashin/pydantic-ddd-interface.git
cd pydantic-ddd-interface

# Install dependencies
poetry install

# Run tests
poetry run pytest

# Format code
poetry run black .
poetry run isort .

---

📄 License

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

---

Made with ❤️ for the Python DDD community

Star ⭐ this repo if you find it useful!