{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/migration\/content-pipeline\/src\/db_interface.py",
"content": "\"\"\"\nDatabase Interface - Abstract base for database operations.\n\nSOLID Principles:\n- Interface Segregation: Separate protocols for different concerns\n- Dependency Inversion: Depend on abstractions, not concrete Database class\n- Single Responsibility: Each protocol has one purpose\n\"\"\"\n\nfrom typing import Protocol, runtime_checkable\n\nfrom enums import DocumentStatus, QueueStatus, SemanticStatus\n\n\n@runtime_checkable\nclass DatabaseConnection(Protocol):\n \"\"\"Protocol for database connection management.\"\"\"\n\n def connect(self) -> bool:\n \"\"\"Establish database connection.\"\"\"\n ...\n\n def disconnect(self) -> None:\n \"\"\"Close database connection.\"\"\"\n ...\n\n def execute(self, query: str, params: tuple = None):\n \"\"\"Execute query and return cursor.\"\"\"\n ...\n\n def commit(self) -> None:\n \"\"\"Commit current transaction.\"\"\"\n ...\n\n def is_connected(self) -> bool:\n \"\"\"Check if connected.\"\"\"\n ...\n\n\n@runtime_checkable\nclass DocumentRepository(Protocol):\n \"\"\"Protocol for document CRUD operations.\"\"\"\n\n def document_exists(self, file_path: str) -> int | None:\n \"\"\"Check if document exists, return ID or None.\"\"\"\n ...\n\n def document_is_done(self, file_path: str) -> int | None:\n \"\"\"Check if document is fully processed.\"\"\"\n ...\n\n def insert_document(\n self,\n file_path: str,\n title: str,\n file_type: str,\n file_size: int,\n file_hash: str,\n ) -> int | None:\n \"\"\"Insert or update document.\"\"\"\n ...\n\n def update_document_status(\n self,\n doc_id: int,\n status: str | DocumentStatus,\n error_message: str = None,\n ) -> None:\n \"\"\"Update document status.\"\"\"\n ...\n\n\n@runtime_checkable\nclass ChunkRepository(Protocol):\n \"\"\"Protocol for chunk operations.\"\"\"\n\n def insert_chunk(\n self,\n doc_id: int,\n chunk_index: int,\n content: str,\n heading_path: str,\n position_start: int = None,\n position_end: int = None,\n metadata: str = None,\n page_id: int = None,\n ) -> int:\n \"\"\"Insert a text chunk.\"\"\"\n ...\n\n def get_chunks_for_embedding(self, limit: int = 100) -> list:\n \"\"\"Get chunks that need embeddings.\"\"\"\n ...\n\n def update_chunk_qdrant_id(self, chunk_id: int, qdrant_id: str) -> None:\n \"\"\"Update chunk with Qdrant point ID.\"\"\"\n ...\n\n\n@runtime_checkable\nclass QueueRepository(Protocol):\n \"\"\"Protocol for pipeline queue operations.\"\"\"\n\n def add_to_queue(self, file_path: str, priority: int = 5) -> int:\n \"\"\"Add file to processing queue.\"\"\"\n ...\n\n def get_pending_queue_items(self, limit: int = 10) -> list:\n \"\"\"Get pending items from queue.\"\"\"\n ...\n\n def update_queue_status(\n self,\n queue_id: int,\n status: str | QueueStatus,\n error_message: str = None,\n ) -> None:\n \"\"\"Update queue item status.\"\"\"\n ...\n\n\n@runtime_checkable\nclass LoggingRepository(Protocol):\n \"\"\"Protocol for logging operations.\"\"\"\n\n def log(self, level: str, message: str, context: str = None) -> None:\n \"\"\"Log a message.\"\"\"\n ...\n\n\n@runtime_checkable\nclass PromptRepository(Protocol):\n \"\"\"Protocol for prompt retrieval.\"\"\"\n\n def get_prompt(self, name: str) -> str | None:\n \"\"\"Get prompt template by name.\"\"\"\n ...\n\n\n@runtime_checkable\nclass SemanticRepository(Protocol):\n \"\"\"Protocol for semantic operations.\"\"\"\n\n def get_entity_types(self) -> list[dict]:\n \"\"\"Get all entity types.\"\"\"\n ...\n\n def get_stopwords(self) -> set[str]:\n \"\"\"Get stopwords set.\"\"\"\n ...\n\n\nclass DatabaseInterface(\n DatabaseConnection,\n DocumentRepository,\n ChunkRepository,\n QueueRepository,\n LoggingRepository,\n PromptRepository,\n SemanticRepository,\n Protocol,\n):\n \"\"\"Combined database interface.\n\n Use this as the type hint for components that need\n the full database functionality.\n \"\"\"\n\n pass\n\n\n# Type aliases for common use cases\nReadOnlyDatabase = DatabaseConnection # Minimal interface\nDocumentDatabase = DocumentRepository # Document operations only\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/migration\/content-pipeline\/src\/db_interface.py",
"content": "\"\"\"\nDatabase Interface - Abstract base for database operations.\n\nSOLID Principles:\n- Interface Segregation: Separate protocols for different concerns\n- Dependency Inversion: Depend on abstractions, not concrete Database class\n- Single Responsibility: Each protocol has one purpose\n\"\"\"\n\nfrom typing import Protocol, runtime_checkable\n\nfrom enums import DocumentStatus, QueueStatus, SemanticStatus\n\n\n@runtime_checkable\nclass DatabaseConnection(Protocol):\n \"\"\"Protocol for database connection management.\"\"\"\n\n def connect(self) -> bool:\n \"\"\"Establish database connection.\"\"\"\n ...\n\n def disconnect(self) -> None:\n \"\"\"Close database connection.\"\"\"\n ...\n\n def execute(self, query: str, params: tuple = None):\n \"\"\"Execute query and return cursor.\"\"\"\n ...\n\n def commit(self) -> None:\n \"\"\"Commit current transaction.\"\"\"\n ...\n\n def is_connected(self) -> bool:\n \"\"\"Check if connected.\"\"\"\n ...\n\n\n@runtime_checkable\nclass DocumentRepository(Protocol):\n \"\"\"Protocol for document CRUD operations.\"\"\"\n\n def document_exists(self, file_path: str) -> int | None:\n \"\"\"Check if document exists, return ID or None.\"\"\"\n ...\n\n def document_is_done(self, file_path: str) -> int | None:\n \"\"\"Check if document is fully processed.\"\"\"\n ...\n\n def insert_document(\n self,\n file_path: str,\n title: str,\n file_type: str,\n file_size: int,\n file_hash: str,\n ) -> int | None:\n \"\"\"Insert or update document.\"\"\"\n ...\n\n def update_document_status(\n self,\n doc_id: int,\n status: str | DocumentStatus,\n error_message: str = None,\n ) -> None:\n \"\"\"Update document status.\"\"\"\n ...\n\n\n@runtime_checkable\nclass ChunkRepository(Protocol):\n \"\"\"Protocol for chunk operations.\"\"\"\n\n def insert_chunk(\n self,\n doc_id: int,\n chunk_index: int,\n content: str,\n heading_path: str,\n position_start: int = None,\n position_end: int = None,\n metadata: str = None,\n page_id: int = None,\n ) -> int:\n \"\"\"Insert a text chunk.\"\"\"\n ...\n\n def get_chunks_for_embedding(self, limit: int = 100) -> list:\n \"\"\"Get chunks that need embeddings.\"\"\"\n ...\n\n def update_chunk_qdrant_id(self, chunk_id: int, qdrant_id: str) -> None:\n \"\"\"Update chunk with Qdrant point ID.\"\"\"\n ...\n\n\n@runtime_checkable\nclass QueueRepository(Protocol):\n \"\"\"Protocol for pipeline queue operations.\"\"\"\n\n def add_to_queue(self, file_path: str, priority: int = 5) -> int:\n \"\"\"Add file to processing queue.\"\"\"\n ...\n\n def get_pending_queue_items(self, limit: int = 10) -> list:\n \"\"\"Get pending items from queue.\"\"\"\n ...\n\n def update_queue_status(\n self,\n queue_id: int,\n status: str | QueueStatus,\n error_message: str = None,\n ) -> None:\n \"\"\"Update queue item status.\"\"\"\n ...\n\n\n@runtime_checkable\nclass LoggingRepository(Protocol):\n \"\"\"Protocol for logging operations.\"\"\"\n\n def log(self, level: str, message: str, context: str = None) -> None:\n \"\"\"Log a message.\"\"\"\n ...\n\n\n@runtime_checkable\nclass PromptRepository(Protocol):\n \"\"\"Protocol for prompt retrieval.\"\"\"\n\n def get_prompt(self, name: str) -> str | None:\n \"\"\"Get prompt template by name.\"\"\"\n ...\n\n\n@runtime_checkable\nclass SemanticRepository(Protocol):\n \"\"\"Protocol for semantic operations.\"\"\"\n\n def get_entity_types(self) -> list[dict]:\n \"\"\"Get all entity types.\"\"\"\n ...\n\n def get_stopwords(self) -> set[str]:\n \"\"\"Get stopwords set.\"\"\"\n ...\n\n\nclass DatabaseInterface(\n DatabaseConnection,\n DocumentRepository,\n ChunkRepository,\n QueueRepository,\n LoggingRepository,\n PromptRepository,\n SemanticRepository,\n Protocol,\n):\n \"\"\"Combined database interface.\n\n Use this as the type hint for components that need\n the full database functionality.\n \"\"\"\n\n pass\n\n\n# Type aliases for common use cases\nReadOnlyDatabase = DatabaseConnection # Minimal interface\nDocumentDatabase = DocumentRepository # Document operations only\n",
"structuredPatch": [],
"originalFile": null
}
}