الحقن بالاعتماد: طريقة بايثون

from typing import Protocol
from abc import ABC, abstractmethod

# تعريف بروتوكول للمخزن
class UserRepository(Protocol):
    def find_by_id(self, user_id: int) -> 'User | None':
        ...
    
    def save(self, user: 'User') -> 'User':
        ...

# الخدمة تعتمد على بروتوكول المخزن
class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo
    
    def get_user(self, user_id: int) -> 'User | None':
        return self.repo.find_by_id(user_id)

class EmailService(Protocol):
    def send(self, to: str, subject: str, body: str) -> None:
        ...

class Logger(Protocol):
    def info(self, msg: str) -> None:
        ...
    
    def error(self, msg: str, err: Exception) -> None:
        ...

class OrderService:
    def __init__(
        self,
        repo: OrderRepository,
        email_svc: EmailService,
        logger: Logger,
        payment_svc: PaymentService,
    ):
        self.repo = repo
        self.email_svc = email_svc
        self.logger = logger
        self.payment_svc = payment_svc

from typing import Protocol

class PaymentProcessor(Protocol):
    def process_payment(self, amount: float) -> bool:
        ...

# أي فئة تحتوي على طريقة process_payment تحقق هذا البروتوكول
class CreditCardProcessor:
    def process_payment(self, amount: float) -> bool:
        # منطق بطاقة الائتمان
        return True

class PayPalProcessor:
    def process_payment(self, amount: float) -> bool:
        # منطق باي بال
        return True

# الخدمة تقبل أي PaymentProcessor
class OrderService:
    def __init__(self, payment_processor: PaymentProcessor):
        self.payment_processor = payment_processor

from abc import ABC, abstractmethod

class PaymentProcessor(ABC):
    @abstractmethod
    def process_payment(self, amount: float) -> bool:
        pass

class CreditCardProcessor(PaymentProcessor):
    def process_payment(self, amount: float) -> bool:
        return True

from typing import Protocol, Optional
from contextlib import contextmanager

class Database(Protocol):
    @contextmanager
    def transaction(self):
        ...
    
    def execute(self, query: str, params: dict) -> None:
        ...
    
    def fetch_one(self, query: str, params: dict) -> Optional[dict]:
        ...

# المخزن يعتمد على المجرة
class UserRepository:
    def __init__(self, db: Database):
        self.db = db
    
    def find_by_id(self, user_id: int) -> Optional['User']:
        result = self.db.fetch_one(
            "SELECT * FROM users WHERE id = :id",
            {"id": user_id}
        )
        if result:
            return User(**result)
        return None

def create_app() -> FastAPI:
    app = FastAPI()
    
    # تهيئة الاعتماديات الأساسية
    db = init_database()
    logger = init_logger()
    
    # تهيئة المخازن
    user_repo = UserRepository(db)
    order_repo = OrderRepository(db)
    
    # تهيئة الخدمة مع الاعتماديات
    email_svc = EmailService(logger)
    payment_svc = PaymentService(logger)
    user_svc = UserService(user_repo, logger)
    order_svc = OrderService(order_repo, email_svc, logger, payment_svc)
    
    # تهيئة معالجات HTTP
    user_handler = UserHandler(user_svc)
    order_handler = OrderHandler(order_svc)
    
    # ربط الطرق
    app.include_router(user_handler.router)
    app.include_router(order_handler.router)
    
    return app

pip install dependency-injector

from dependency_injector import containers, providers
from dependency_injector.wiring import inject, Provide

class Container(containers.DeclarativeContainer):
    # التكوين
    config = providers.Configuration()
    
    # قاعدة البيانات
    db = providers.Singleton(
        Database,
        connection_string=config.database.url
    )
    
    # المخازن
    user_repository = providers.Factory(
        UserRepository,
        db=db
    )
    
    # الخدمة
    user_service = providers.Factory(
        UserService,
        repo=user_repository
    )

# الاستخدام
container = Container()
container.config.database.url.from_env("DATABASE_URL")
user_service = container.user_service()

pip install injector

from injector import Injector, inject, Module, provider

class DatabaseModule(Module):
    @provider
    def provide_db(self) -> Database:
        return Database(connection_string="...")

class UserModule(Module):
    @inject
    def __init__(self, repo: UserRepository):
        self.repo = repo

injector = Injector([DatabaseModule()])
user_service = injector.get(UserService)

from unittest.mock import Mock
import pytest

# تنفيذ مُحاكي للاختبار
class MockUserRepository:
    def __init__(self):
        self.users = {}
        self.error = None
    
    def find_by_id(self, user_id: int) -> Optional['User']:
        if self.error:
            raise self.error
        return self.users.get(user_id)
    
    def save(self, user: 'User') -> 'User':
        if self.error:
            raise self.error
        self.users[user.id] = user
        return user

# اختبار باستخدام المُحاكي
def test_user_service_get_user():
    mock_repo = MockUserRepository()
    mock_repo.users[1] = User(id=1, name="John", email="john@example.com")
    
    service = UserService(mock_repo)
    
    user = service.get_user(1)
    assert user is not None
    assert user.name == "John"

@pytest.fixture
def mock_user_repository():
    return MockUserRepository()

@pytest.fixture
def user_service(mock_user_repository):
    return UserService(mock_user_repository)

def test_user_service_get_user(user_service, mock_user_repository):
    user = User(id=1, name="John", email="john@example.com")
    mock_user_repository.users[1] = user
    
    result = user_service.get_user(1)
    assert result.name == "John"

# جيد: العميل يحتاج فقط إلى قراءة المستخدمين
class UserReader(Protocol):
    def find_by_id(self, user_id: int) -> Optional['User']:
        ...
    
    def find_by_email(self, email: str) -> Optional['User']:
        ...

# واجهة منفصلة للكتابة
class UserWriter(Protocol):
    def save(self, user: 'User') -> 'User':
        ...
    
    def delete(self, user_id: int) -> bool:
        ...

class UserService:
    def __init__(self, repo: UserRepository):
        if repo is None:
            raise ValueError("user repository cannot be None")
        self.repo = repo

from typing import Protocol, Optional

class UserService:
    def __init__(
        self,
        repo: UserRepository,
        logger: Logger,
        email_service: EmailService,
    ) -> None:
        self.repo = repo
        self.logger = logger
        self.email_service = email_service

# جيد: لا يحتاج المُساعِد الداخلي إلى الحقول المُستخدمة
class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo
        # ذاكرة ت高速 - لا تحتاج إلى الحقول المُستخدمة
        self._cache: dict[int, User] = {}

class UserService:
    """UserService يتعامل مع منطق العمل المرتبط بالمستخدمين.
    
    Args:
        repo: UserRepository للاستعلام عن البيانات. يجب أن يكون آمنًا من التحديد
            إذا تم استخدامه في السياقات المتزامنة.
        logger: Logger لتعقب الأخطاء وتصحيحها.
    """
    def __init__(self, repo: UserRepository, logger: Logger):
        self.repo = repo
        self.logger = logger

from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

def get_user_repository() -> UserRepository:
    db = get_database()
    return UserRepository(db)

def get_user_service(
    repo: Annotated[UserRepository, Depends(get_user_repository)]
) -> UserService:
    return UserService(repo)

@app.get("/users/{user_id}")
def get_user(
    user_id: int,
    service: Annotated[UserService, Depends(get_user_service)]
):
    user = service.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404)
    return user

# فئة بيانات بسيطة - لا حاجة للحقول المُستخدمة
from dataclasses import dataclass

@dataclass
class Point:
    x: float
    y: float

# أداة بسيطة - لا حاجة للحقول المُستخدمة
def format_currency(amount: float) -> str:
    return f"${amount:.2f}"

class ReportService:
    def __init__(
        self,
        pdf_generator: PDFGenerator,
        repo: ReportRepository,
        logger: Logger,
    ):
        self.pdf_generator = pdf_generator
        self.repo = repo
        self.logger = logger
    
    def generate_report(self, report_id: int) -> bytes:
        report_data = self.repo.get_by_id(report_id)
        pdf = self.pdf_generator.generate(report_data)
        self.logger.info(f"تم إنشاء التقرير {report_id}")
        return pdf

from typing import Protocol
import asyncio

class AsyncUserRepository(Protocol):
    async def find_by_id(self, user_id: int) -> Optional['User']:
        ...
    
    async def save(self, user: 'User') -> 'User':
        ...

class AsyncUserService:
    def __init__(self, repo: AsyncUserRepository):
        self.repo = repo
    
    async def get_user(self, user_id: int) -> Optional['User']:
        return await self.repo.find_by_id(user_id)
    
    async def get_users_batch(self, user_ids: list[int]) -> list['User']:
        tasks = [self.repo.find_by_id(uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
        return [u for u in results if u is not None]