أنماط تصميم بايثون لعمارة نظيفة

# سيء: مسؤوليات متعددة
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

# استخدام البروتوكول (Python 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:
        # منطق باي بال
        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:
        # التحقق من وجود المستخدم
        existing_user = self.user_repository.get_by_email(email)
        if existing_user:
            raise UserAlreadyExistsError(f"المستخدم ذو البريد الإلكتروني {email} موجود بالفعل")
        
        # إنشاء مستخدم جديد
        user = User(
            id=uuid4(),
            email=email,
            name=name,
            is_active=True
        )
        
        # حفظ في المستودع
        user = self.user_repository.save(user)
        
        # إرسال بريد ترحيب
        self.email_service.send(
            to=user.email,
            subject="مرحبا!",
            body=f"مرحبا {user.name}، ترحيبًا بكم على منصتنا!"
        )
        
        # نشر الحدث
        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_id} غير موجود")
        
        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):
        # استخدام الاعتماديات المحقونة
        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"لا يوجد تسجيل متوفر لـ {interface}")

# الاستخدام
def create_container() -> Container:
    container = Container()
    
    # تسجيل الخدمات
    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

# port المدخل (الأساسي)
class CreateUserUseCase(Protocol):
    def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
        ...

# port المخرج (الثانوي)
class UserPersistencePort(Protocol):
    def save(self, user: User) -> User:
        ...
    
    def find_by_email(self, email: str) -> Optional[User]:
        ...

from pydantic import BaseModel, EmailStr

# adapter المدخل (واجهة REST)
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))

# adapter المخرج (قاعدة البيانات)
# تم تنفيذه بالفعل كـ 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"بريد إلكتروني غير صحيح: {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("لا يمكن أن يكون المبلغ سالبًا")
        if self.currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError(f"عملة غير مدعومة: {self.currency}")
    
    def add(self, other: 'Money') -> 'Money':
        if self.currency != other.currency:
            raise ValueError("لا يمكن إضافة عملات مختلفة")
        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("لا يمكن تأكيد طلب فارغ")
        if self.status != "pending":
            raise ValueError("الطلب تم معالجته بالفعل")
        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('لا يمكن أن تحتوي الاسم على أرقام')
        return v
    
    class Config:
        frozen = True  # جعله غير قابل للتغيير

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/           # طبقة واجهة المستخدم/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():
    # التحضير
    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)
    
    # التنفيذ
    user = service.register_user("test@example.com", "Test")
    
    # التأكيد
    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  # كيان نظيف