Dependency Injector example

notes.md

Step 1: Define Service Classes and Interfaces

# services.py
import requests

class CsvContentFetcher:
    def fetch(self, csv_uri: str) -> str:
        # Depending on the scheme in `csv_uri`, fetch the content from S3 or local file
        if csv_uri.startswith("s3://"):
            # Assume `fetch_from_s3` is a utility function you've implemented
            return self.fetch_from_s3(csv_uri)
        else:
            # Local file handling
            with open(csv_uri, "r") as file:
                return file.read()

    def fetch_from_s3(self, s3_uri: str) -> str:
        # Mock implementation for fetching content from S3
        return "name,units\community A,10"

class MduAuthServiceClient:
    def __init__(self, base_url: str):
        self.base_url = base_url

    def create_community(self, community_data: str):
        print(f"Creating community with data:\n{community_data}\nusing base URL: {self.base_url}")

class CommunityCreator:
    def __init__(self, auth_service: MduAuthServiceClient, csv_content: str, validation_criteria: dict):
        self.auth_service = auth_service
        self.csv_content = csv_content
        self.validation_criteria = validation_criteria

    def create(self):
        # Here, use `self.csv_content` directly for community creation
        self.auth_service.create_community(self.csv_content)
        # Post-creation validation logic...

Step 2: Setup Configuration and Container

# containers.py
from dependency_injector import containers, providers
from services import CsvContentFetcher, MduAuthServiceClient, CommunityCreator

class AppConfig(containers.DeclarativeContainer):
    config = providers.Configuration()

class Container(containers.DeclarativeContainer):
    configuration = providers.Container(AppConfig)

    csv_content_fetcher = providers.Singleton(CsvContentFetcher)

    mdu_auth_service_client = providers.Singleton(
        MduAuthServiceClient,
        base_url=configuration.config.mdu_auth_service.base_url,
    )

    community_creator = providers.Factory(
        CommunityCreator,
        auth_service=mdu_auth_service_client,
        csv_content=providers.Callable(lambda csv_uri: csv_content_fetcher().fetch(csv_uri),
                                       csv_uri=configuration.config.community.csv_uri),
        validation_criteria=configuration.config.community.validation_criteria,
    )

Step 3: Main Application Logic with Command-Line Argument Overrides

# main.py
import argparse
from containers import Container

def main():
    parser = argparse.ArgumentParser(description="Community Creator Tool")
    parser.add_argument("--mdu-auth-service-url", type=str, help="MDU Auth Service Base URL")
    parser.add_argument("--csv-uri", type=str, help="URI to the community CSV file")
    args = parser.parse_args()

    # Setting up the container with possible overrides from command-line arguments
    container = Container()
    if args.mdu_auth_service_url:
        container.configuration.config.mdu_auth_service.base_url.override(args.mdu_auth_service_url)
    if args.csv_uri:
        container.configuration.config.community.csv_uri.override(args.csv_uri)

    community_creator = container.community_creator()
    community_creator.create()

if __name__ == "__main__":
    main()

Step 4: Unit Testing with fake HTTP Server

# fake_mdu_auth_server.py
from http.server import BaseHTTPRequestHandler, HTTPServer
import threading
import json

class FakeMduAuthHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path == "/create_community":
            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            response = {"status": "success", "message": "Community created"}
            self.wfile.write(json.dumps(response).encode('utf-8'))
        else:
            self.send_response(404)
    def log_message(self, format, *args):
        return  # Suppress log messages

class StoppableHTTPServer(HTTPServer):
    def run(self):
        try:
            self.serve_forever()
        except KeyboardInterrupt:
            pass
        finally:
            self.server_close()

def run_fake_server(port=8000):
    server = StoppableHTTPServer(('', port), FakeMduAuthHandler)
    server_thread = threading.Thread(target=server.run)
    server_thread.start()
    return server, server_thread

then test_community_creator.py

import pytest
from fake_mdu_auth_server import run_fake_server
from containers import Container

@pytest.fixture(scope="session", autouse=True)
def fake_mdu_auth_server():
    server, server_thread = run_fake_server(8000)
    yield
    server.shutdown()
    server_thread.join()

@pytest.fixture(scope="function")
def override_mdu_auth_service_url():
    container = Container()
    container.config.mdu_auth_service.base_url.override("http://localhost:8000")
    container.config.from_yaml("config_test.yaml")  # Assuming you have a test-specific config
    container.wire(modules=["__main__"])
    yield container
    container.unwire()

def test_community_creation(override_mdu_auth_service_url):
    # Assuming your Container and CommunityCreator are properly set up to use the overridden URL
    community_creator = override_mdu_auth_service_url.community_creator()
    # Proceed with using community_creator for testing
    # This is where you would invoke methods on community_creator and assert expected outcomes