📘 Domain-Driven Design: Python Implementation

🎯 Introduction

Welcome to this exciting tutorial on Domain-Driven Design (DDD) in Python! 🎉 In this guide, we’ll explore how to build software that truly reflects your business domain.

You’ll discover how DDD can transform your Python development experience. Whether you’re building microservices 🌐, enterprise applications 🖥️, or complex business systems 📚, understanding DDD is essential for writing robust, maintainable code that speaks your business language.

By the end of this tutorial, you’ll feel confident implementing DDD patterns in your own projects! Let’s dive in! 🏊‍♂️

📚 Understanding Domain-Driven Design

🤔 What is Domain-Driven Design?

Domain-Driven Design is like building a house with blueprints that everyone can understand 🏗️. Think of it as creating software that speaks the same language as your business experts - no translation needed!

In Python terms, DDD is a strategic approach to software design that focuses on modeling software to match a domain. This means you can:

• ✨ Create code that business experts can understand
• 🚀 Build systems that evolve with business needs
• 🛡️ Protect business logic from technical concerns

💡 Why Use DDD?

Here’s why developers love DDD:

1. Ubiquitous Language 🗣️: Everyone speaks the same language
2. Bounded Contexts 📦: Clear boundaries between different parts
3. Rich Domain Models 💎: Business logic lives where it belongs
4. Event-Driven Architecture 🎯: Systems that react to business events

Real-world example: Imagine building an e-commerce platform 🛒. With DDD, your code would use terms like “Order”, “Customer”, and “Inventory” - exactly what the business uses!

🔧 Basic Syntax and Usage

📝 Simple Example: Building Blocks

Let’s start with DDD’s core building blocks:

# 👋 Hello, DDD!
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
import uuid

# 🎨 Value Object - Immutable and defined by its attributes
@dataclass(frozen=True)
class Money:
    amount: float
    currency: str = "USD"  # 💵 Default currency
    
    def __add__(self, other: 'Money') -> 'Money':
        if self.currency != other.currency:
            raise ValueError("Cannot add different currencies! 💔")
        return Money(self.amount + other.amount, self.currency)

# 🏗️ Entity - Has identity and lifecycle
class Product:
    def __init__(self, name: str, price: Money):
        self.id = str(uuid.uuid4())  # 🔑 Unique identity
        self.name = name
        self.price = price
        self.stock = 0
        
    def add_stock(self, quantity: int):
        """📦 Add items to inventory"""
        self.stock += quantity
        print(f"Added {quantity} {self.name}(s) to stock! 📈")

💡 Explanation: Notice how we use domain terms directly! Money is a value object (immutable), while Product is an entity (has identity).

🎯 Domain Events

Here’s how we capture things that happen in our domain:

# 🎯 Domain Event - Something that happened
@dataclass
class ProductPurchased:
    product_id: str
    quantity: int
    price: Money
    purchased_at: datetime
    customer_id: str
    
    def __str__(self):
        return f"🛍️ Customer {self.customer_id} bought {self.quantity} items!"

# 🚀 Aggregate - Consistency boundary
class ShoppingCart:
    def __init__(self, customer_id: str):
        self.id = str(uuid.uuid4())
        self.customer_id = customer_id
        self.items: List[tuple[Product, int]] = []
        self.events: List[ProductPurchased] = []  # 📝 Event sourcing!
        
    def add_item(self, product: Product, quantity: int):
        """🛒 Add product to cart"""
        if product.stock < quantity:
            raise ValueError(f"Not enough stock! Only {product.stock} available 😢")
            
        self.items.append((product, quantity))
        print(f"Added {quantity} {product.name}(s) to cart! ✨")
        
    def checkout(self) -> List[ProductPurchased]:
        """💳 Complete the purchase"""
        events = []
        for product, quantity in self.items:
            event = ProductPurchased(
                product_id=product.id,
                quantity=quantity,
                price=Money(product.price.amount * quantity),
                purchased_at=datetime.now(),
                customer_id=self.customer_id
            )
            events.append(event)
            product.stock -= quantity  # 📉 Update inventory
            
        self.events.extend(events)
        self.items.clear()  # 🧹 Empty the cart
        return events

💡 Practical Examples

🏦 Example 1: Banking Domain

Let’s build a simple banking system:

# 💰 Banking domain model
from enum import Enum
from decimal import Decimal

class AccountType(Enum):
    CHECKING = "checking"
    SAVINGS = "savings"
    
@dataclass(frozen=True)
class AccountNumber:
    """🔢 Value object for account numbers"""
    value: str
    
    def __post_init__(self):
        if not self.value or len(self.value) != 10:
            raise ValueError("Account number must be 10 digits! 🚫")

class BankAccount:
    """🏦 Bank account aggregate"""
    def __init__(self, account_number: AccountNumber, account_type: AccountType):
        self.account_number = account_number
        self.account_type = account_type
        self.balance = Decimal("0.00")
        self.is_active = True
        self.transactions: List[Transaction] = []
        
    def deposit(self, amount: Decimal) -> 'Transaction':
        """💵 Deposit money"""
        if amount <= 0:
            raise ValueError("Amount must be positive! 💸")
            
        if not self.is_active:
            raise ValueError("Account is closed! 🔒")
            
        self.balance += amount
        transaction = Transaction(
            account_number=self.account_number,
            amount=amount,
            transaction_type="DEPOSIT",
            balance_after=self.balance
        )
        self.transactions.append(transaction)
        print(f"✅ Deposited ${amount}! New balance: ${self.balance}")
        return transaction
        
    def withdraw(self, amount: Decimal) -> 'Transaction':
        """💸 Withdraw money"""
        if amount > self.balance:
            raise ValueError(f"Insufficient funds! Only ${self.balance} available 😅")
            
        self.balance -= amount
        transaction = Transaction(
            account_number=self.account_number,
            amount=amount,
            transaction_type="WITHDRAWAL",
            balance_after=self.balance
        )
        self.transactions.append(transaction)
        print(f"💳 Withdrew ${amount}! Remaining: ${self.balance}")
        return transaction

@dataclass
class Transaction:
    """📝 Domain event for transactions"""
    account_number: AccountNumber
    amount: Decimal
    transaction_type: str
    balance_after: Decimal
    timestamp: datetime = datetime.now()
    
# 🎮 Let's use our banking domain!
account = BankAccount(
    AccountNumber("1234567890"),
    AccountType.CHECKING
)
account.deposit(Decimal("1000.00"))
account.withdraw(Decimal("50.00"))

🎯 Try it yourself: Add a transfer method that moves money between accounts!

🚗 Example 2: Ride-Sharing Domain

Let’s model a ride-sharing service:

# 🚗 Ride-sharing domain
from geopy.distance import geodesic

@dataclass(frozen=True)
class Location:
    """📍 Value object for locations"""
    latitude: float
    longitude: float
    address: str
    
    def distance_to(self, other: 'Location') -> float:
        """📏 Calculate distance in kilometers"""
        return geodesic(
            (self.latitude, self.longitude),
            (other.latitude, other.longitude)
        ).km

class RideStatus(Enum):
    REQUESTED = "🔵 Requested"
    DRIVER_ASSIGNED = "🟡 Driver Assigned"
    IN_PROGRESS = "🟢 In Progress"
    COMPLETED = "✅ Completed"
    CANCELLED = "❌ Cancelled"

class Ride:
    """🚗 Ride aggregate"""
    def __init__(self, rider_id: str, pickup: Location, destination: Location):
        self.id = str(uuid.uuid4())
        self.rider_id = rider_id
        self.driver_id: Optional[str] = None
        self.pickup = pickup
        self.destination = destination
        self.status = RideStatus.REQUESTED
        self.price: Optional[Money] = None
        self.events: List[dict] = []
        
        # 💰 Calculate estimated price
        distance = pickup.distance_to(destination)
        self.estimated_price = Money(
            amount=round(5.0 + (distance * 2.5), 2)  # Base + per km
        )
        
        self._record_event("RideRequested", {
            "pickup": pickup.address,
            "destination": destination.address,
            "estimated_price": self.estimated_price.amount
        })
        
    def assign_driver(self, driver_id: str):
        """🚗 Assign a driver to the ride"""
        if self.status != RideStatus.REQUESTED:
            raise ValueError(f"Cannot assign driver in {self.status.value} status! 🚫")
            
        self.driver_id = driver_id
        self.status = RideStatus.DRIVER_ASSIGNED
        self._record_event("DriverAssigned", {"driver_id": driver_id})
        print(f"🎯 Driver {driver_id} assigned to ride!")
        
    def start_ride(self):
        """🏁 Start the ride"""
        if self.status != RideStatus.DRIVER_ASSIGNED:
            raise ValueError("No driver assigned yet! 🤷")
            
        self.status = RideStatus.IN_PROGRESS
        self._record_event("RideStarted", {"start_time": datetime.now()})
        print(f"🚗 Ride started! Heading to {self.destination.address}")
        
    def complete_ride(self, actual_distance: float):
        """🏁 Complete the ride"""
        if self.status != RideStatus.IN_PROGRESS:
            raise ValueError("Ride not in progress! 🛑")
            
        # 💵 Calculate final price based on actual distance
        self.price = Money(amount=round(5.0 + (actual_distance * 2.5), 2))
        self.status = RideStatus.COMPLETED
        
        self._record_event("RideCompleted", {
            "final_price": self.price.amount,
            "distance": actual_distance
        })
        print(f"✅ Ride completed! Total: ${self.price.amount}")
        
    def _record_event(self, event_type: str, data: dict):
        """📝 Record domain events"""
        self.events.append({
            "type": event_type,
            "timestamp": datetime.now(),
            "data": data
        })

# 🎮 Test our ride-sharing domain!
pickup = Location(40.7128, -74.0060, "Times Square, NYC 🗽")
destination = Location(40.7614, -73.9776, "Central Park, NYC 🌳")

ride = Ride("customer123", pickup, destination)
print(f"💰 Estimated price: ${ride.estimated_price.amount}")

ride.assign_driver("driver456")
ride.start_ride()
ride.complete_ride(actual_distance=3.2)

🚀 Advanced Concepts

🧙‍♂️ Repository Pattern

When you’re ready to level up, implement the repository pattern:

# 🎯 Repository pattern for data access
from abc import ABC, abstractmethod
from typing import Dict, Optional, List

class Repository(ABC):
    """📦 Abstract repository interface"""
    
    @abstractmethod
    def add(self, entity):
        pass
        
    @abstractmethod
    def get(self, entity_id: str):
        pass
        
    @abstractmethod
    def update(self, entity):
        pass
        
    @abstractmethod
    def delete(self, entity_id: str):
        pass

class InMemoryProductRepository(Repository):
    """💾 In-memory implementation for testing"""
    
    def __init__(self):
        self.storage: Dict[str, Product] = {}
        
    def add(self, product: Product):
        """➕ Add product to repository"""
        self.storage[product.id] = product
        print(f"💾 Saved product: {product.name}")
        
    def get(self, product_id: str) -> Optional[Product]:
        """🔍 Find product by ID"""
        return self.storage.get(product_id)
        
    def update(self, product: Product):
        """🔄 Update existing product"""
        if product.id not in self.storage:
            raise ValueError(f"Product {product.id} not found! 😱")
        self.storage[product.id] = product
        
    def delete(self, product_id: str):
        """🗑️ Remove product"""
        if product_id in self.storage:
            del self.storage[product_id]
            print(f"🗑️ Deleted product {product_id}")
            
    def find_by_name(self, name: str) -> List[Product]:
        """🔎 Custom query method"""
        return [p for p in self.storage.values() if name.lower() in p.name.lower()]

🏗️ Domain Services

For complex business logic that doesn’t belong to a single entity:

# 🚀 Domain service for complex operations
class PricingService:
    """💰 Service for pricing calculations"""
    
    def __init__(self, discount_repository: Repository):
        self.discount_repository = discount_repository
        
    def calculate_total(self, items: List[tuple[Product, int]], 
                       customer_type: str = "regular") -> Money:
        """🧮 Calculate total with discounts"""
        subtotal = Money(0.0)
        
        for product, quantity in items:
            item_total = Money(product.price.amount * quantity)
            subtotal = subtotal + item_total
            
        # 🎁 Apply customer discount
        discount_rate = self._get_discount_rate(customer_type)
        if discount_rate > 0:
            discount_amount = Money(subtotal.amount * discount_rate)
            total = Money(subtotal.amount - discount_amount.amount)
            print(f"🎉 Applied {discount_rate*100}% discount!")
            return total
            
        return subtotal
        
    def _get_discount_rate(self, customer_type: str) -> float:
        """🏷️ Get discount rate by customer type"""
        discounts = {
            "regular": 0.0,
            "premium": 0.1,   # 10% off
            "vip": 0.2        # 20% off
        }
        return discounts.get(customer_type, 0.0)

⚠️ Common Pitfalls and Solutions

😱 Pitfall 1: Anemic Domain Models

# ❌ Wrong way - no behavior, just data!
class Order:
    def __init__(self):
        self.id = None
        self.items = []
        self.total = 0.0
        self.status = "pending"

# ✅ Correct way - rich domain model with behavior!
class Order:
    def __init__(self, customer_id: str):
        self.id = str(uuid.uuid4())
        self.customer_id = customer_id
        self.items: List[OrderItem] = []
        self.status = OrderStatus.PENDING
        
    def add_item(self, product: Product, quantity: int):
        """🛒 Business logic lives here!"""
        if self.status != OrderStatus.PENDING:
            raise ValueError("Cannot modify confirmed order! 🔒")
        # More business rules...
        
    def calculate_total(self) -> Money:
        """💰 Order knows how to calculate its total"""
        return sum(item.subtotal() for item in self.items)

🤯 Pitfall 2: Mixing Infrastructure with Domain

# ❌ Dangerous - domain depends on infrastructure!
class Product:
    def save_to_database(self):
        import sqlite3  # 💥 Domain shouldn't know about DB!
        conn = sqlite3.connect('products.db')
        # ...

# ✅ Safe - keep infrastructure separate!
class Product:
    """Pure domain model"""
    def update_price(self, new_price: Money):
        if new_price.amount < 0:
            raise ValueError("Price cannot be negative! 💸")
        self.price = new_price
        
# Infrastructure layer handles persistence
class SQLProductRepository(Repository):
    def add(self, product: Product):
        # Database logic here
        pass

🛠️ Best Practices

1. 🎯 Ubiquitous Language: Use the same terms as your domain experts
2. 📦 Bounded Contexts: Keep related concepts together, separate others
3. 🛡️ Protect Invariants: Aggregates enforce business rules
4. 🎨 Rich Domain Models: Put behavior where the data is
5. ✨ Event Sourcing: Consider storing events instead of state

🧪 Hands-On Exercise

🎯 Challenge: Build a Hotel Booking System

Create a DDD-based hotel booking system:

📋 Requirements:

• ✅ Rooms with different types (single, double, suite)
• 🏷️ Dynamic pricing based on season and occupancy
• 👤 Guest profiles with loyalty levels
• 📅 Booking with check-in/check-out dates
• 🎨 Each room type needs special amenities!

🚀 Bonus Points:

• Add event sourcing for bookings
• Implement overbooking protection
• Create a pricing strategy pattern

💡 Solution

# 🎯 Hotel booking domain model!
from enum import Enum
from datetime import date, timedelta

class RoomType(Enum):
    SINGLE = ("single", "🛏️", 100.0)
    DOUBLE = ("double", "🛏️🛏️", 150.0)
    SUITE = ("suite", "🏰", 300.0)
    
    def __init__(self, value, emoji, base_price):
        self._value_ = value
        self.emoji = emoji
        self.base_price = base_price

class LoyaltyLevel(Enum):
    BRONZE = ("bronze", "🥉", 0.05)
    SILVER = ("silver", "🥈", 0.10)
    GOLD = ("gold", "🥇", 0.15)
    
    def __init__(self, value, emoji, discount):
        self._value_ = value
        self.emoji = emoji
        self.discount = discount

@dataclass(frozen=True)
class DateRange:
    """📅 Value object for date ranges"""
    check_in: date
    check_out: date
    
    def __post_init__(self):
        if self.check_out <= self.check_in:
            raise ValueError("Check-out must be after check-in! 📆")
            
    @property
    def nights(self) -> int:
        return (self.check_out - self.check_in).days
        
    def overlaps_with(self, other: 'DateRange') -> bool:
        """🔍 Check if date ranges overlap"""
        return not (self.check_out <= other.check_in or 
                   self.check_in >= other.check_out)

class Room:
    """🏨 Room entity"""
    def __init__(self, room_number: str, room_type: RoomType):
        self.room_number = room_number
        self.room_type = room_type
        self.bookings: List['Booking'] = []
        self.amenities = self._get_amenities()
        
    def _get_amenities(self) -> List[str]:
        """✨ Get amenities by room type"""
        amenities_map = {
            RoomType.SINGLE: ["📺 TV", "☕ Coffee maker"],
            RoomType.DOUBLE: ["📺 TV", "☕ Coffee maker", "🧊 Mini-bar"],
            RoomType.SUITE: ["📺 TV", "☕ Coffee maker", "🧊 Mini-bar", 
                            "🛁 Jacuzzi", "🍾 Champagne"]
        }
        return amenities_map[self.room_type]
        
    def is_available(self, date_range: DateRange) -> bool:
        """🔍 Check if room is available"""
        for booking in self.bookings:
            if booking.status == BookingStatus.CONFIRMED and \
               booking.date_range.overlaps_with(date_range):
                return False
        return True

class BookingStatus(Enum):
    PENDING = "🔵 Pending"
    CONFIRMED = "✅ Confirmed"
    CANCELLED = "❌ Cancelled"
    CHECKED_IN = "🏨 Checked In"
    CHECKED_OUT = "👋 Checked Out"

class Booking:
    """📋 Booking aggregate"""
    def __init__(self, guest_id: str, room: Room, date_range: DateRange):
        if not room.is_available(date_range):
            raise ValueError(f"Room {room.room_number} not available! 😢")
            
        self.id = str(uuid.uuid4())
        self.guest_id = guest_id
        self.room = room
        self.date_range = date_range
        self.status = BookingStatus.PENDING
        self.total_price: Optional[Money] = None
        self.events: List[dict] = []
        
        self._record_event("BookingCreated", {
            "room": room.room_number,
            "dates": f"{date_range.check_in} to {date_range.check_out}"
        })
        
    def confirm(self, pricing_service: 'PricingService', guest: 'Guest'):
        """✅ Confirm the booking"""
        if self.status != BookingStatus.PENDING:
            raise ValueError("Can only confirm pending bookings! 🚫")
            
        self.total_price = pricing_service.calculate_booking_price(
            self.room, self.date_range, guest.loyalty_level
        )
        self.status = BookingStatus.CONFIRMED
        self.room.bookings.append(self)
        
        self._record_event("BookingConfirmed", {
            "price": self.total_price.amount,
            "loyalty_discount": guest.loyalty_level.discount
        })
        
        print(f"✅ Booking confirmed! {self.room.room_type.emoji}")
        print(f"💰 Total: ${self.total_price.amount} for {self.date_range.nights} nights")
        
    def _record_event(self, event_type: str, data: dict):
        self.events.append({
            "type": event_type,
            "timestamp": datetime.now(),
            "data": data
        })

class Guest:
    """👤 Guest entity"""
    def __init__(self, name: str, email: str):
        self.id = str(uuid.uuid4())
        self.name = name
        self.email = email
        self.loyalty_level = LoyaltyLevel.BRONZE
        self.total_nights = 0
        
    def update_loyalty_status(self):
        """🏆 Update loyalty based on nights stayed"""
        if self.total_nights >= 50:
            self.loyalty_level = LoyaltyLevel.GOLD
            print(f"🎉 Congratulations! You're now {self.loyalty_level.emoji} level!")
        elif self.total_nights >= 20:
            self.loyalty_level = LoyaltyLevel.SILVER
            print(f"🎊 Great! You've reached {self.loyalty_level.emoji} level!")

class HotelPricingService:
    """💰 Domain service for pricing"""
    def calculate_booking_price(self, room: Room, date_range: DateRange, 
                               loyalty_level: LoyaltyLevel) -> Money:
        """🧮 Calculate total price with all factors"""
        base_price = room.room_type.base_price
        nights = date_range.nights
        
        # 📈 Seasonal pricing (summer is peak)
        if date_range.check_in.month in [6, 7, 8]:
            base_price *= 1.5  # 50% increase
            print("☀️ Peak season pricing applied!")
            
        subtotal = base_price * nights
        
        # 🎁 Apply loyalty discount
        discount = subtotal * loyalty_level.discount
        total = subtotal - discount
        
        if discount > 0:
            print(f"{loyalty_level.emoji} Loyalty discount: ${discount:.2f}")
            
        return Money(round(total, 2))

# 🎮 Test our hotel system!
hotel_room = Room("101", RoomType.SUITE)
guest = Guest("Alice Johnson", "[email protected]")
guest.total_nights = 25  # Silver member!
guest.update_loyalty_status()

dates = DateRange(
    date.today() + timedelta(days=30),
    date.today() + timedelta(days=33)
)

booking = Booking(guest.id, hotel_room, dates)
pricing_service = HotelPricingService()
booking.confirm(pricing_service, guest)

print(f"\n🏨 Room amenities: {', '.join(hotel_room.amenities)}")

🎓 Key Takeaways

You’ve learned so much! Here’s what you can now do:

• ✅ Create rich domain models with confidence 💪
• ✅ Implement DDD building blocks (entities, value objects, aggregates) 🛡️
• ✅ Apply ubiquitous language in real projects 🎯
• ✅ Design bounded contexts like a pro 🐛
• ✅ Build event-driven systems with Python! 🚀

Remember: DDD is about modeling your software to match your business domain. It’s here to help you write better, more maintainable code! 🤝

🤝 Next Steps

Congratulations! 🎉 You’ve mastered Domain-Driven Design in Python!

Here’s what to do next:

1. 💻 Practice with the hotel booking exercise above
2. 🏗️ Apply DDD to your current project’s most complex domain
3. 📚 Move on to our next tutorial: Event Sourcing and CQRS
4. 🌟 Share your DDD journey with others!

Remember: Every DDD expert was once a beginner. Keep modeling, keep learning, and most importantly, have fun building systems that truly reflect your business! 🚀

Happy coding! 🎉🚀✨