साफ़ आर्किटेक्चर के लिए पाइथन डिज़ाइन पैटर्न

# बुरा: कई जिम्मेदारियाँ
class UserManager:
    def create_user(self, user_data):
        # यूज़र बनाना
        pass

    def send_welcome_email(self, user):
        # ईमेल भेजना
        pass

    def log_creation(self, user):
        # फाइल में लॉग करना
        pass

# अच्छा: अलग-अलग जिम्मेदारियाँ
class UserService:
    def __init__(self, repository, email_service, logger):
        self.repository = repository
        self.email_service = email_service
        self.logger = logger

    def create_user(self, user_data):
        user = User(**user_data)
        self.repository.save(user)
        self.email_service.send_welcome(user)
        self.logger.info(f"यूज़र बनाया गया: {user.id}")
        return user

from abc import ABC, abstractmethod
from typing import Protocol

# प्रोटोकॉल का उपयोग (पाइथन 3.8+)
class PaymentProcessor(Protocol):
    def process_payment(self, amount: float) -> bool:
        ...

class CreditCardProcessor:
    def process_payment(self, amount: float) -> bool:
        # क्रेडिट कार्ड लॉजिक
        return True

class PayPalProcessor:
    def process_payment(self, amount: float) -> bool:
        # PayPal लॉजिक
        return True

# मौजूदा कोड को मॉडिफाई किए बिना आसानी से एक्सटेंसिबल
class CryptoProcessor:
    def process_payment(self, amount: float) -> bool:
        # क्रिप्टोक्यूरेंसी लॉजिक
        return True

from abc import ABC, abstractmethod

class DataStore(ABC):
    @abstractmethod
    def save(self, key: str, value: str) -> None:
        pass

    @abstractmethod
    def get(self, key: str) -> str:
        pass

class PostgreSQLStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # PostgreSQL इम्प्लीमेंटेशन
        pass

    def get(self, key: str) -> str:
        # PostgreSQL इम्प्लीमेंटेशन
        return ""

class RedisStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # Redis इम्प्लीमेंटेशन
        pass

    def get(self, key: str) -> str:
        # Redis इम्प्लीमेंटेशन
        return ""

# दोनों को इंटरचेंजेबल रूप से उपयोग किया जा सकता है
def process_data(store: DataStore, key: str, value: str):
    store.save(key, value)
    return store.get(key)

# बुरा: फैट इंटरफेस
class Worker(ABC):
    @abstractmethod
    def work(self): pass

    @abstractmethod
    def eat(self): pass

    @abstractmethod
    def sleep(self): pass

# अच्छा: सेग्रेगेटेड इंटरफेस
class Workable(Protocol):
    def work(self) -> None: ...

class Eatable(Protocol):
    def eat(self) -> None: ...

class Human:
    def work(self) -> None:
        print("काम कर रहा हूँ")

    def eat(self) -> None:
        print("खा रहा हूँ")

class Robot:
    def work(self) -> None:
        print("काम कर रहा हूँ")
    # ईट मेथड की आवश्यकता नहीं है

from typing import Protocol

# अभिसरण
class EmailSender(Protocol):
    def send(self, to: str, subject: str, body: str) -> None:
        ...

# लो-लेवल मॉड्यूल
class SMTPEmailSender:
    def send(self, to: str, subject: str, body: str) -> None:
        # SMTP इम्प्लीमेंटेशन
        pass

# हाई-लेवल मॉड्यूल अभिसरण पर निर्भर है
class UserRegistrationService:
    def __init__(self, email_sender: EmailSender):
        self.email_sender = email_sender

    def register(self, email: str, name: str):
        # रजिस्ट्रेशन लॉजिक
        self.email_sender.send(
            to=email,
            subject="स्वागत है!",
            body=f"नमस्ते {name}"
        )

from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4

@dataclass
class User:
    id: UUID
    email: str
    name: str
    is_active: bool = True

class UserRepository(ABC):
    @abstractmethod
    def save(self, user: User) -> User:
        pass

    @abstractmethod
    def get_by_id(self, user_id: UUID) -> Optional[User]:
        pass

    @abstractmethod
    def get_by_email(self, email: str) -> Optional[User]:
        pass

    @abstractmethod
    def list_all(self) -> List[User]:
        pass

    @abstractmethod
    def delete(self, user_id: UUID) -> bool:
        pass

from sqlalchemy import create_engine, Column, String, Boolean
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

Base = declarative_base()

class UserModel(Base):
    __tablename__ = 'users'

    id = Column(PGUUID(as_uuid=True), primary_key=True)
    email = Column(String, unique=True, nullable=False)
    name = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)

class SQLAlchemyUserRepository(UserRepository):
    def __init__(self, session: Session):
        self.session = session

    def save(self, user: User) -> User:
        user_model = UserModel(
            id=user.id,
            email=user.email,
            name=user.name,
            is_active=user.is_active
        )
        self.session.add(user_model)
        self.session.commit()
        return user

    def get_by_id(self, user_id: UUID) -> Optional[User]:
        user_model = self.session.query(UserModel).filter(
            UserModel.id == user_id
        ).first()

        if not user_model:
            return None

        return User(
            id=user_model.id,
            email=user_model.email,
            name=user_model.name,
            is_active=user_model.is_active
        )

    def get_by_email(self, email: str) -> Optional[User]:
        user_model = self.session.query(UserModel).filter(
            UserModel.email == email
        ).first()

        if not user_model:
            return None

        return User(
            id=user_model.id,
            email=user_model.email,
            name=user_model.name,
            is_active=user_model.is_active
        )

    def list_all(self) -> List[User]:
        users = self.session.query(UserModel).all()
        return [
            User(
                id=u.id,
                email=u.email,
                name=u.name,
                is_active=u.is_active
            )
            for u in users
        ]

    def delete(self, user_id: UUID) -> bool:
        result = self.session.query(UserModel).filter(
            UserModel.id == user_id
        ).delete()
        self.session.commit()
        return result > 0

class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self.users: dict[UUID, User] = {}

    def save(self, user: User) -> User:
        self.users[user.id] = user
        return user

    def get_by_id(self, user_id: UUID) -> Optional[User]:
        return self.users.get(user_id)

    def get_by_email(self, email: str) -> Optional[User]:
        for user in self.users.values():
            if user.email == email:
                return user
        return None

    def list_all(self) -> List[User]:
        return list(self.users.values())

    def delete(self, user_id: UUID) -> bool:
        if user_id in self.users:
            del self.users[user_id]
            return True
        return False

from typing import Optional
from uuid import uuid4

class UserAlreadyExistsError(Exception):
    pass

class UserNotFoundError(Exception):
    pass

class UserService:
    def __init__(
        self,
        user_repository: UserRepository,
        email_service: EmailSender,
        event_publisher: 'EventPublisher'
    ):
        self.user_repository = user_repository
        self.email_service = email_service
        self.event_publisher = event_publisher

    def register_user(self, email: str, name: str) -> User:
        # Check if user exists
        existing_user = self.user_repository.get_by_email(email)
        if existing_user:
            raise UserAlreadyExistsError(f"User with email {email} already exists")

        # Create new user
        user = User(
            id=uuid4(),
            email=email,
            name=name,
            is_active=True
        )

        # Save to repository
        user = self.user_repository.save(user)

        # Send welcome email
        self.email_service.send(
            to=user.email,
            subject="Welcome!",
            body=f"Hello {user.name}, welcome to our platform!"
        )

        # Publish event
        self.event_publisher.publish('user.registered', {
            'user_id': str(user.id),
            'email': user.email
        })

        return user

    def deactivate_user(self, user_id: UUID) -> User:
        user = self.user_repository.get_by_id(user_id)
        if not user:
            raise UserNotFoundError(f"User {user_id} not found")

        user.is_active = False
        user = self.user_repository.save(user)

        self.event_publisher.publish('user.deactivated', {
            'user_id': str(user.id)
        })

        return user

class OrderService:
    def __init__(
        self,
        order_repository: 'OrderRepository',
        payment_processor: PaymentProcessor,
        notification_service: 'NotificationService'
    ):
        self.order_repository = order_repository
        self.payment_processor = payment_processor
        self.notification_service = notification_service

    def place_order(self, order_data: dict):
        # Use injected dependencies
        pass

from typing import Dict, Type, Callable, Any

class Container:
    def __init__(self):
        self._services: Dict[Type, Callable] = {}
        self._singletons: Dict[Type, Any] = {}

    def register(self, interface: Type, factory: Callable):
        self._services[interface] = factory

    def register_singleton(self, interface: Type, instance: Any):
        self._singletons[interface] = instance

    def resolve(self, interface: Type):
        if interface in self._singletons:
            return self._singletons[interface]

        factory = self._services.get(interface)
        if factory:
            return factory(self)

        raise ValueError(f"No registration found for {interface}")

# Usage
def create_container() -> Container:
    container = Container()

    # Register services
    container.register_singleton(
        Session,
        sessionmaker(bind=create_engine('postgresql://...'))()
    )

    container.register(
        UserRepository,
        lambda c: SQLAlchemyUserRepository(c.resolve(Session))
    )

    container.register(
        EmailSender,
        lambda c: SMTPEmailSender()
    )

    container.register(
        UserService,
        lambda c: UserService(
            c.resolve(UserRepository),
            c.resolve(EmailSender),
            c.resolve(EventPublisher)
        )
    )

    return container

# Input Port (Primary)
class CreateUserUseCase(Protocol):
    def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
        ...

# Output Port (Secondary)
class UserPersistencePort(Protocol):
    def save(self, user: User) -> User:
        ...

    def find_by_email(self, email: str) -> Optional[User]:
        ...

from pydantic import BaseModel, EmailStr

# Input Adapter (REST API)
from fastapi import FastAPI, Depends, HTTPException

class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str

class CreateUserResponse(BaseModel):
    id: str
    email: str
    name: str

app = FastAPI()

@app.post("/users", response_model=CreateUserResponse)
def create_user(
    request: CreateUserRequest,
    user_service: UserService = Depends(get_user_service)
):
    try:
        user = user_service.register_user(
            email=request.email,
            name=request.name
        )
        return CreateUserResponse(
            id=str(user.id),
            email=user.email,
            name=user.name
        )
    except UserAlreadyExistsError as e:
        raise HTTPException(status_code=400, detail=str(e))

# Output Adapter (Database)
# Already implemented as SQLAlchemyUserRepository

from dataclasses import dataclass
from typing import Pattern
import re

@dataclass(frozen=True)
class Email:
    value: str

    EMAIL_PATTERN: Pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')

    def __post_init__(self):
        if not self.EMAIL_PATTERN.match(self.value):
            raise ValueError(f"Invalid email: {self.value}")

    def __str__(self):
        return self.value

@dataclass(frozen=True)
class Money:
    amount: float
    currency: str

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Amount cannot be negative")
        if self.currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError(f"Unsupported currency: {self.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)

from dataclasses import dataclass, field
from typing import List
from datetime import datetime

@dataclass
class OrderItem:
    product_id: UUID
    quantity: int
    price: Money

    def total(self) -> Money:
        return Money(
            self.price.amount * self.quantity,
            self.price.currency
        )

@dataclass
class Order:
    id: UUID
    customer_id: UUID
    items: List[OrderItem] = field(default_factory=list)
    status: str = "pending"
    created_at: datetime = field(default_factory=datetime.now)

    def add_item(self, product_id: UUID, quantity: int, price: Money):
        item = OrderItem(product_id, quantity, price)
        self.items.append(item)

    def remove_item(self, product_id: UUID):
        self.items = [
            item for item in self.items
            if item.product_id != product_id
        ]

    def total(self) -> Money:
        if not self.items:
            return Money(0, "USD")

        return sum(
            (item.total() for item in self.items),
            Money(0, self.items[0].price.currency)
        )

    def confirm(self):
        if not self.items:
            raise ValueError("Cannot confirm empty order")
        if self.status != "pending":
            raise ValueError("Order already processed")
        self.status = "confirmed"

from dataclasses import dataclass
from datetime import datetime
from typing import List, Callable

@dataclass
class DomainEvent:
    occurred_at: datetime = field(default_factory=datetime.now)

@dataclass
class OrderConfirmed(DomainEvent):
    order_id: UUID
    customer_id: UUID
    total: Money

class EventPublisher:
    def __init__(self):
        self._handlers: Dict[Type, List[Callable]] = {}

    def subscribe(self, event_type: Type, handler: Callable):
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)

    def publish(self, event: DomainEvent):
        event_type = type(event)
        handlers = self._handlers.get(event_type, [])
        for handler in handlers:
            handler(event)

from typing import Protocol, runtime_checkable

@runtime_checkable
class Serializable(Protocol):
    def to_dict(self) -> dict:
        ...

    @classmethod
    def from_dict(cls, data: dict) -> 'Serializable':
        ...

def serialize(obj: Serializable) -> dict:
    return obj.to_dict()

from pydantic import BaseModel, Field, validator
from typing import Optional

class CreateUserDTO(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2, max_length=100)
    age: Optional[int] = Field(None, ge=0, le=150)

    @validator('name')
    def name_must_not_contain_numbers(cls, v):
        if any(char.isdigit() for char in v):
            raise ValueError('Name cannot contain numbers')
        return v

    class Config:
        frozen = True  # Make immutable

from typing import List
import asyncio

class AsyncUserRepository(ABC):
    @abstractmethod
    async def save(self, user: User) -> User:
        pass

    @abstractmethod
    async def get_by_id(self, user_id: UUID) -> Optional[User]:
        pass

class AsyncUserService:
    def __init__(self, repository: AsyncUserRepository):
        self.repository = repository

    async def register_user(self, email: str, name: str) -> User:
        user = User(id=uuid4(), email=email, name=name)
        return await self.repository.save(user)

    async def get_users_batch(self, user_ids: List[UUID]) -> List[User]:
        tasks = [self.repository.get_by_id(uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
        return [u for u in results if u is not None]

my_application/
├── domain/                 # एंटरप्राइज बिजनेस रूल्स
│   ├── __init__.py
│   ├── entities/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── order.py
│   ├── value_objects/
│   │   ├── __init__.py
│   │   ├── email.py
│   │   └── money.py
│   ├── events/
│   │   ├── __init__.py
│   │   └── user_events.py
│   └── exceptions.py
│
├── application/            # एप्लिकेशन बिजनेस रूल्स
│   ├── __init__.py
│   ├── use_cases/
│   │   ├── __init__.py
│   │   ├── create_user.py
│   │   └── place_order.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── ports/
│       ├── __init__.py
│       ├── repositories.py
│       └── external_services.py
│
├── infrastructure/         # बाहरी इंटरफेस
│   ├── __init__.py
│   ├── persistence/
│   │   ├── __init__.py
│   │   ├── sqlalchemy/
│   │   │   ├── models.py
│   │   │   └── repositories.py
│   │   └── mongodb/
│   │       └── repositories.py
│   ├── messaging/
│   │   ├── __init__.py
│   │   └── rabbitmq_publisher.py
│   ├── external_services/
│   │   ├── __init__.py
│   │   └── email_service.py
│   └── config.py
│
├── presentation/           # UI/API लेयर
│   ├── __init__.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── dependencies.py
│   │   ├── routes/
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   └── orders.py
│   │   └── schemas/
│   │       ├── __init__.py
│   │       └── user_schemas.py
│   └── cli/
│       └── commands.py
│
├── tests/
│   ├── unit/
│   ├── integration/
│   └── e2e/
│
├── main.py                 # एप्लिकेशन एंट्री पॉइंट
├── container.py            # डिपेंडेंसी इंजेक्शन सेटअप
├── pyproject.toml
└── README.md

import pytest
from uuid import uuid4

def test_user_creation():
    user = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )
    assert user.email == "test@example.com"
    assert user.is_active is True

def test_order_total_calculation():
    order = Order(id=uuid4(), customer_id=uuid4())
    order.add_item(
        uuid4(),
        quantity=2,
        price=Money(10.0, "USD")
    )
    order.add_item(
        uuid4(),
        quantity=1,
        price=Money(5.0, "USD")
    )
    assert order.total().amount == 25.0

@pytest.fixture
def in_memory_repository():
    return InMemoryUserRepository()

def test_user_repository_save_and_retrieve(in_memory_repository):
    user = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )

    saved_user = in_memory_repository.save(user)
    retrieved_user = in_memory_repository.get_by_id(user.id)

    assert retrieved_user is not None
    assert retrieved_user.email == user.email

from unittest.mock import Mock

def test_user_registration():
    # Arrange
    mock_repository = Mock(spec=UserRepository)
    mock_repository.get_by_email.return_value = None
    mock_repository.save.return_value = User(
        id=uuid4(),
        email="test@example.com",
        name="Test"
    )

    mock_email = Mock(spec=EmailSender)
    mock_events = Mock(spec=EventPublisher)

    service = UserService(mock_repository, mock_email, mock_events)

    # Act
    user = service.register_user("test@example.com", "Test")

    # Assert
    assert user.email == "test@example.com"
    mock_repository.save.assert_called_once()
    mock_email.send.assert_called_once()
    mock_events.publish.assert_called_once()

# बुरा
from sqlalchemy import Column

@dataclass
class User:
    id: Column(Integer, primary_key=True)  # फ्रेमवर्क डोमेन में लीक हो रहा है

# अच्छा
@dataclass
class User:
    id: UUID  # शुद्ध डोमेन ऑब्जेक्ट