आवलंबन एंजेक्शन: एक पायथन तरीका

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="जॉन", email="जॉन@example.com")
    
    service = UserService(mock_repo)
    
    user = service.get_user(1)
    assert user is not None
    assert user.name == "जॉन"

@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="जॉन", email="जॉन@example.com")
    mock_user_repository.users[1] = user
    
    result = user_service.get_user(1)
    assert result.name == "जॉन"

# अच्छा: ग्राहक केवल उपयोगकर्ता पढ़ने के लिए आवश्यकता है
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("उपयोगकर्ता रिपॉज़िटरी शून्य नहीं हो सकता")
        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 उपयोगकर्ता संबंधी व्यवसाय तर्क का प्रबंधन करता है।
    
    तर्क:
        repo: UserRepository डेटा एक्सेस के लिए। यदि समानांतर संदर्भ में उपयोग किया जाता है, तो यह धागा-सुरक्षित होना चाहिए
        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]