{
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/mcp-servers\/mcp_docs\/infrastructure\/docs_repository.py",
"content": "\"\"\"Repository für Dokumentation CRUD-Operationen\"\"\"\nimport sys\nfrom datetime import datetime\nfrom typing import Optional, List, Dict, Any\n\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_docs\")\nfrom config import Config\nfrom domain.dokumentation import Dokumentation, DocStatus\nfrom infrastructure.db_connection import DatabaseConnection\n\n\nclass DocsRepository:\n \"\"\"Repository für ki_system.dokumentation Tabelle\"\"\"\n\n def find_by_id(self, doc_id: int) -> Optional[Dokumentation]:\n \"\"\"Findet Dokument nach ID\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n cursor.execute(\n \"SELECT * FROM dokumentation WHERE id = %s\",\n (doc_id,)\n )\n row = cursor.fetchone()\n return self._row_to_doc(row) if row else None\n\n def find_by_path(self, path: str) -> Optional[Dokumentation]:\n \"\"\"Findet Dokument nach Pfad\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n cursor.execute(\n \"SELECT * FROM dokumentation WHERE path = %s\",\n (path,)\n )\n row = cursor.fetchone()\n return self._row_to_doc(row) if row else None\n\n def find_by_slug(self, slug: str, parent_id: Optional[int] = None) -> Optional[Dokumentation]:\n \"\"\"Findet Dokument nach Slug (optional mit Parent)\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n if parent_id is not None:\n cursor.execute(\n \"SELECT * FROM dokumentation WHERE slug = %s AND parent_id = %s\",\n (slug, parent_id)\n )\n else:\n cursor.execute(\n \"SELECT * FROM dokumentation WHERE slug = %s AND parent_id IS NULL\",\n (slug,)\n )\n row = cursor.fetchone()\n return self._row_to_doc(row) if row else None\n\n def find_all(\n self,\n status: Optional[str] = None,\n parent_id: Optional[int] = None,\n search: Optional[str] = None,\n limit: int = 20,\n offset: int = 0\n ) -> List[Dokumentation]:\n \"\"\"Findet alle Dokumente mit optionalen Filtern\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n conditions = []\n params = []\n\n if status:\n conditions.append(\"status = %s\")\n params.append(status)\n\n if parent_id is not None:\n conditions.append(\"parent_id = %s\")\n params.append(parent_id)\n\n if search:\n conditions.append(\"(title LIKE %s OR description LIKE %s OR content LIKE %s)\")\n search_term = f\"%{search}%\"\n params.extend([search_term, search_term, search_term])\n\n where_clause = \" AND \".join(conditions) if conditions else \"1=1\"\n\n # Limit begrenzen\n limit = min(limit, Config.MAX_RESULTS)\n\n sql = f\"\"\"\n SELECT * FROM dokumentation\n WHERE {where_clause}\n ORDER BY depth, sort_order, title\n LIMIT %s OFFSET %s\n \"\"\"\n params.extend([limit, offset])\n\n cursor.execute(sql, params)\n rows = cursor.fetchall()\n return [self._row_to_doc(row) for row in rows]\n\n def find_children(self, parent_id: int) -> List[Dokumentation]:\n \"\"\"Findet alle direkten Kinder eines Dokuments\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n cursor.execute(\n \"\"\"SELECT * FROM dokumentation\n WHERE parent_id = %s\n ORDER BY sort_order, title\"\"\",\n (parent_id,)\n )\n rows = cursor.fetchall()\n return [self._row_to_doc(row) for row in rows]\n\n def find_root_documents(self) -> List[Dokumentation]:\n \"\"\"Findet alle Root-Dokumente (ohne Parent)\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n cursor.execute(\n \"\"\"SELECT * FROM dokumentation\n WHERE parent_id IS NULL\n ORDER BY sort_order, title\"\"\"\n )\n rows = cursor.fetchall()\n return [self._row_to_doc(row) for row in rows]\n\n def get_hierarchy(self) -> List[Dict[str, Any]]:\n \"\"\"Gibt kompletten Dokumentationsbaum zurück\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n cursor.execute(\n \"\"\"SELECT id, parent_id, slug, path, title, status, sort_order, depth\n FROM dokumentation\n ORDER BY depth, sort_order, title\"\"\"\n )\n rows = cursor.fetchall()\n\n # Baum aufbauen\n docs_by_id = {row[\"id\"]: {**row, \"children\": []} for row in rows}\n roots = []\n\n for row in rows:\n doc = docs_by_id[row[\"id\"]]\n if row[\"parent_id\"] is None:\n roots.append(doc)\n elif row[\"parent_id\"] in docs_by_id:\n docs_by_id[row[\"parent_id\"]][\"children\"].append(doc)\n\n return roots\n\n def get_breadcrumb(self, doc_id: int) -> List[Dict[str, Any]]:\n \"\"\"Gibt Breadcrumb-Pfad für ein Dokument zurück\"\"\"\n breadcrumb = []\n current_id = doc_id\n\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n while current_id is not None:\n cursor.execute(\n \"SELECT id, parent_id, path, title FROM dokumentation WHERE id = %s\",\n (current_id,)\n )\n row = cursor.fetchone()\n if row:\n breadcrumb.insert(0, {\n \"id\": row[\"id\"],\n \"path\": row[\"path\"],\n \"title\": row[\"title\"]\n })\n current_id = row[\"parent_id\"]\n else:\n break\n\n return breadcrumb\n\n def get_siblings(self, doc_id: int) -> List[Dokumentation]:\n \"\"\"Findet Geschwister-Dokumente (gleicher Parent)\"\"\"\n doc = self.find_by_id(doc_id)\n if not doc:\n return []\n\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n if doc.parent_id is not None:\n cursor.execute(\n \"\"\"SELECT * FROM dokumentation\n WHERE parent_id = %s AND id != %s\n ORDER BY sort_order, title\"\"\",\n (doc.parent_id, doc_id)\n )\n else:\n cursor.execute(\n \"\"\"SELECT * FROM dokumentation\n WHERE parent_id IS NULL AND id != %s\n ORDER BY sort_order, title\"\"\",\n (doc_id,)\n )\n rows = cursor.fetchall()\n return [self._row_to_doc(row) for row in rows]\n\n def count(\n self,\n status: Optional[str] = None,\n parent_id: Optional[int] = None,\n search: Optional[str] = None\n ) -> int:\n \"\"\"Zählt Dokumente mit optionalen Filtern\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n conditions = []\n params = []\n\n if status:\n conditions.append(\"status = %s\")\n params.append(status)\n\n if parent_id is not None:\n conditions.append(\"parent_id = %s\")\n params.append(parent_id)\n\n if search:\n conditions.append(\"(title LIKE %s OR description LIKE %s OR content LIKE %s)\")\n search_term = f\"%{search}%\"\n params.extend([search_term, search_term, search_term])\n\n where_clause = \" AND \".join(conditions) if conditions else \"1=1\"\n\n cursor.execute(f\"SELECT COUNT(*) as count FROM dokumentation WHERE {where_clause}\", params)\n result = cursor.fetchone()\n return result[\"count\"] if result else 0\n\n def create(self, doc: Dokumentation) -> int:\n \"\"\"Erstellt neues Dokument, gibt ID zurück\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n now = datetime.now()\n cursor.execute(\n \"\"\"INSERT INTO dokumentation\n (parent_id, slug, path, title, description, content, status, sort_order, depth, created_at, updated_at)\n VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)\"\"\",\n (\n doc.parent_id,\n doc.slug,\n doc.path,\n doc.title,\n doc.description,\n doc.content,\n doc.status.value if isinstance(doc.status, DocStatus) else doc.status,\n doc.sort_order,\n doc.depth,\n now,\n now\n )\n )\n return cursor.lastrowid\n\n def update(self, doc_id: int, updates: Dict[str, Any]) -> bool:\n \"\"\"Aktualisiert Dokument-Felder\"\"\"\n if not updates:\n ... [TRUNCATED-a7ad1062d286074e]",
"numLines": 385,
"startLine": 1,
"totalLines": 385
}
}
}