{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/migration\/content-pipeline\/src\/container.py",
"content": "\"\"\"\nDependency Injection Container.\n\nSOLID: Dependency Inversion - Central place for wiring dependencies.\nReplaces global instances with managed dependencies.\n\nUsage:\n from container import Container\n\n # Get database instance (singleton)\n db = Container.get_database()\n\n # Get LLM provider\n llm = Container.get_llm_provider()\n\n # Create step with injected dependencies\n step = Container.create_embedding_step()\n\"\"\"\n\nfrom typing import Any, TypeVar\n\nfrom db_interface import DatabaseInterface\nfrom enums import LLMProvider\n\nT = TypeVar(\"T\")\n\n\nclass Container:\n \"\"\"Simple dependency injection container.\n\n Manages singleton instances and creates components with\n proper dependency injection.\n \"\"\"\n\n _instances: dict[str, Any] = {}\n _factories: dict[str, callable] = {}\n\n @classmethod\n def reset(cls) -> None:\n \"\"\"Reset container (for testing).\"\"\"\n cls._instances.clear()\n\n @classmethod\n def register(cls, name: str, factory: callable) -> None:\n \"\"\"Register a factory function for a dependency.\n\n Args:\n name: Dependency name\n factory: Factory function that creates the instance\n \"\"\"\n cls._factories[name] = factory\n\n @classmethod\n def get(cls, name: str) -> Any:\n \"\"\"Get or create a singleton instance.\n\n Args:\n name: Dependency name\n\n Returns:\n Singleton instance\n \"\"\"\n if name not in cls._instances:\n if name in cls._factories:\n cls._instances[name] = cls._factories[name]()\n else:\n raise KeyError(f\"No factory registered for: {name}\")\n return cls._instances[name]\n\n @classmethod\n def get_database(cls) -> DatabaseInterface:\n \"\"\"Get database instance (singleton).\n\n Returns:\n Database instance implementing DatabaseInterface\n \"\"\"\n if \"database\" not in cls._instances:\n from db import Database\n\n cls._instances[\"database\"] = Database()\n return cls._instances[\"database\"]\n\n @classmethod\n def get_llm_provider(\n cls,\n provider: LLMProvider | str | None = None,\n use_anthropic: bool = False,\n model: str | None = None,\n ):\n \"\"\"Get LLM provider instance.\n\n Args:\n provider: Explicit provider type\n use_anthropic: Use Anthropic if True (legacy param)\n model: Optional model override\n\n Returns:\n LLM provider instance\n \"\"\"\n from llm_interface import LLMFactory, get_llm_provider\n\n if provider:\n return LLMFactory.create(provider, model=model)\n return get_llm_provider(use_anthropic=use_anthropic, model=model)\n\n @classmethod\n def create_embedding_step(cls, progress=None):\n \"\"\"Create EmbeddingStep with injected database.\n\n Args:\n progress: Optional PipelineProgress instance\n\n Returns:\n Configured EmbeddingStep\n \"\"\"\n from step_embed import EmbeddingStep\n\n return EmbeddingStep(cls.get_database(), progress)\n\n @classmethod\n def create_extraction_step(cls, progress=None):\n \"\"\"Create ExtractionStep with injected database.\n\n Args:\n progress: Optional PipelineProgress instance\n\n Returns:\n Configured ExtractionStep\n \"\"\"\n from step_extract import ExtractionStep\n\n return ExtractionStep(cls.get_database(), progress)\n\n @classmethod\n def create_load_step(cls, progress=None):\n \"\"\"Create LoadStep with injected database.\n\n Args:\n progress: Optional PipelineProgress instance\n\n Returns:\n Configured LoadStep\n \"\"\"\n from step_load import LoadStep\n\n return LoadStep(cls.get_database(), progress)\n\n @classmethod\n def create_transform_step(cls, progress=None):\n \"\"\"Create TransformationStep with injected database.\n\n Args:\n progress: Optional PipelineProgress instance\n\n Returns:\n Configured TransformationStep\n \"\"\"\n from step_transform import TransformationStep\n\n return TransformationStep(cls.get_database(), progress)\n\n @classmethod\n def create_semantic_step(cls, progress=None):\n \"\"\"Create SemanticStep with injected database.\n\n Args:\n progress: Optional PipelineProgress instance\n\n Returns:\n Configured SemanticStep\n \"\"\"\n from step_semantic import SemanticStep\n\n return SemanticStep(cls.get_database(), progress)\n\n\n# Convenience functions for backward compatibility\ndef get_db() -> DatabaseInterface:\n \"\"\"Get database instance.\n\n Prefer this over importing `db` directly for testability.\n \"\"\"\n return Container.get_database()\n\n\ndef get_llm(use_anthropic: bool = False, model: str | None = None):\n \"\"\"Get LLM provider instance.\n\n Args:\n use_anthropic: Use Anthropic instead of Ollama\n model: Optional model override\n\n Returns:\n LLM provider instance\n \"\"\"\n return Container.get_llm_provider(use_anthropic=use_anthropic, model=model)\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/migration\/content-pipeline\/src\/container.py",
"content": "\"\"\"\nDependency Injection Container.\n\nSOLID: Dependency Inversion - Central place for wiring dependencies.\nReplaces global instances with managed dependencies.\n\nUsage:\n from container import Container\n\n # Get database instance (singleton)\n db = Container.get_database()\n\n # Get LLM provider\n llm = Container.get_llm_provider()\n\n # Create step with injected dependencies\n step = Container.create_embedding_step()\n\"\"\"\n\nfrom typing import Any, TypeVar\n\nfrom db_interface import DatabaseInterface\nfrom enums import LLMProvider\n\nT = TypeVar(\"T\")\n\n\nclass Container:\n \"\"\"Simple dependency injection container.\n\n Manages singleton instances and creates components with\n proper dependency injection.\n \"\"\"\n\n _instances: dict[str, Any] = {}\n _factories: dict[str, callable] = {}\n\n @classmethod\n def reset(cls) -> None:\n \"\"\"Reset container (for testing).\"\"\"\n cls._instances.clear()\n\n @classmethod\n def register(cls, name: str, factory: callable) -> None:\n \"\"\"Register a factory function for a dependency.\n\n Args:\n name: Dependency name\n factory: Factory function that creates the instance\n \"\"\"\n cls._factories[name] = factory\n\n @classmethod\n def get(cls, name: str) -> Any:\n \"\"\"Get or create a singleton instance.\n\n Args:\n name: Dependency name\n\n Returns:\n Singleton instance\n \"\"\"\n if name not in cls._instances:\n if name in cls._factories:\n cls._instances[name] = cls._factories[name]()\n else:\n raise KeyError(f\"No factory registered for: {name}\")\n return cls._instances[name]\n\n @classmethod\n def get_database(cls) -> DatabaseInterface:\n \"\"\"Get database instance (singleton).\n\n Returns:\n Database instance implementing DatabaseInterface\n \"\"\"\n if \"database\" not in cls._instances:\n from db import Database\n\n cls._instances[\"database\"] = Database()\n return cls._instances[\"database\"]\n\n @classmethod\n def get_llm_provider(\n cls,\n provider: LLMProvider | str | None = None,\n use_anthropic: bool = False,\n model: str | None = None,\n ):\n \"\"\"Get LLM provider instance.\n\n Args:\n provider: Explicit provider type\n use_anthropic: Use Anthropic if True (legacy param)\n model: Optional model override\n\n Returns:\n LLM provider instance\n \"\"\"\n from llm_interface import LLMFactory, get_llm_provider\n\n if provider:\n return LLMFactory.create(provider, model=model)\n return get_llm_provider(use_anthropic=use_anthropic, model=model)\n\n @classmethod\n def create_embedding_step(cls, progress=None):\n \"\"\"Create EmbeddingStep with injected database.\n\n Args:\n progress: Optional PipelineProgress instance\n\n Returns:\n Configured EmbeddingStep\n \"\"\"\n from step_embed import EmbeddingStep\n\n return EmbeddingStep(cls.get_database(), progress)\n\n @classmethod\n def create_extraction_step(cls, progress=None):\n \"\"\"Create ExtractionStep with injected database.\n\n Args:\n progress: Optional PipelineProgress instance\n\n Returns:\n Configured ExtractionStep\n \"\"\"\n from step_extract import ExtractionStep\n\n return ExtractionStep(cls.get_database(), progress)\n\n @classmethod\n def create_load_step(cls, progress=None):\n \"\"\"Create LoadStep with injected database.\n\n Args:\n progress: Optional PipelineProgress instance\n\n Returns:\n Configured LoadStep\n \"\"\"\n from step_load import LoadStep\n\n return LoadStep(cls.get_database(), progress)\n\n @classmethod\n def create_transform_step(cls, progress=None):\n \"\"\"Create TransformationStep with injected database.\n\n Args:\n progress: Optional PipelineProgress instance\n\n Returns:\n Configured TransformationStep\n \"\"\"\n from step_transform import TransformationStep\n\n return TransformationStep(cls.get_database(), progress)\n\n @classmethod\n def create_semantic_step(cls, progress=None):\n \"\"\"Create SemanticStep with injected database.\n\n Args:\n progress: Optional PipelineProgress instance\n\n Returns:\n Configured SemanticStep\n \"\"\"\n from step_semantic import SemanticStep\n\n return SemanticStep(cls.get_database(), progress)\n\n\n# Convenience functions for backward compatibility\ndef get_db() -> DatabaseInterface:\n \"\"\"Get database instance.\n\n Prefer this over importing `db` directly for testability.\n \"\"\"\n return Container.get_database()\n\n\ndef get_llm(use_anthropic: bool = False, model: str | None = None):\n \"\"\"Get LLM provider instance.\n\n Args:\n use_anthropic: Use Anthropic instead of Ollama\n model: Optional model override\n\n Returns:\n LLM provider instance\n \"\"\"\n return Container.get_llm_provider(use_anthropic=use_anthropic, model=model)\n",
"structuredPatch": [],
"originalFile": null
}
}