Protokoll #27979
| ID | 27979 |
|---|---|
| Zeitstempel | 2025-12-28 12:38:48.959031 |
| Client | root |
| IP | 145.224.72.140 |
| Modell | claude-sonnet-4-20250514 |
| Status | completed |
| Tokens | 13,117 (Input: 2,657, Output: 10,460) |
| Dauer | 94 ms |
| Request-Zeit | 2025-12-28 12:38:48.959031 |
| Response-Zeit | 2025-12-28 12:38:49.053257 |
Request
{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp_docs\/infrastructure\/docs_repository.py",
"content": "\"\"\"Repository für Dokumentation CRUD-Operationen.\"\"\"\n\nfrom datetime import datetime\nfrom typing import Any, Dict, List, Optional\n\nfrom config import Config\nfrom domain.dokumentation import DocStatus, Dokumentation\nfrom shared.infrastructure import SimpleDbConnection\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 SimpleDbConnection.get_connection(Config) as conn:\n with conn.cursor() as cursor:\n cursor.execute(\n \"SELECT * FROM dokumentation WHERE id = %s\", (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 SimpleDbConnection.get_connection(Config) as conn:\n with conn.cursor() as cursor:\n cursor.execute(\n \"SELECT * FROM dokumentation WHERE path = %s\", (path,)\n )\n row = cursor.fetchone()\n return self._row_to_doc(row) if row else None\n\n def find_by_slug(\n self, slug: str, parent_id: Optional[int] = None\n ) -> Optional[Dokumentation]:\n \"\"\"Findet Dokument nach Slug (optional mit Parent).\"\"\"\n with SimpleDbConnection.get_connection(Config) 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 SimpleDbConnection.get_connection(Config) as conn:\n with conn.cursor() as cursor:\n conditions = []\n params: List[Any] = []\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(\n \"(title LIKE %s OR description LIKE %s OR content LIKE %s)\"\n )\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 SimpleDbConnection.get_connection(Config) 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 SimpleDbConnection.get_connection(Config) 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 SimpleDbConnection.get_connection(Config) 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: List[Dict[str, Any]] = []\n current_id: Optional[int] = doc_id\n\n with SimpleDbConnection.get_connection(Config) 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(\n 0,\n {\n \"id\": row[\"id\"],\n \"path\": row[\"path\"],\n \"title\": row[\"title\"],\n },\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 SimpleDbConnection.get_connection(Config) 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 SimpleDbConnection.get_connection(Config) as conn:\n with conn.cursor() as cursor:\n conditions = []\n params: List[Any] = []\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(\n \"(title LIKE %s OR description LIKE %s OR content LIKE %s)\"\n )\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(\n f\"SELECT COUNT(*) as count FROM dokumentation WHERE {where_clause}\",\n params,\n )\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 SimpleDbConnection.get_connection(Config) 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,\n 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\n if isinstance(doc.status, DocStatus)\n else d... [TRUNCATED-e6e81cf9864f22cf]"
}
}Response
{
"tool_response": {
"type": "update",
"filePath": "\/var\/www\/mcp-servers\/mcp_docs\/infrastructure\/docs_repository.py",
"content": "\"\"\"Repository für Dokumentation CRUD-Operationen.\"\"\"\n\nfrom datetime import datetime\nfrom typing import Any, Dict, List, Optional\n\nfrom config import Config\nfrom domain.dokumentation import DocStatus, Dokumentation\nfrom shared.infrastructure import SimpleDbConnection\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 SimpleDbConnection.get_connection(Config) as conn:\n with conn.cursor() as cursor:\n cursor.execute(\n \"SELECT * FROM dokumentation WHERE id = %s\", (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 SimpleDbConnection.get_connection(Config) as conn:\n with conn.cursor() as cursor:\n cursor.execute(\n \"SELECT * FROM dokumentation WHERE path = %s\", (path,)\n )\n row = cursor.fetchone()\n return self._row_to_doc(row) if row else None\n\n def find_by_slug(\n self, slug: str, parent_id: Optional[int] = None\n ) -> Optional[Dokumentation]:\n \"\"\"Findet Dokument nach Slug (optional mit Parent).\"\"\"\n with SimpleDbConnection.get_connection(Config) 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 SimpleDbConnection.get_connection(Config) as conn:\n with conn.cursor() as cursor:\n conditions = []\n params: List[Any] = []\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(\n \"(title LIKE %s OR description LIKE %s OR content LIKE %s)\"\n )\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 SimpleDbConnection.get_connection(Config) 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 SimpleDbConnection.get_connection(Config) 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 SimpleDbConnection.get_connection(Config) 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: List[Dict[str, Any]] = []\n current_id: Optional[int] = doc_id\n\n with SimpleDbConnection.get_connection(Config) 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(\n 0,\n {\n \"id\": row[\"id\"],\n \"path\": row[\"path\"],\n \"title\": row[\"title\"],\n },\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 SimpleDbConnection.get_connection(Config) 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 SimpleDbConnection.get_connection(Config) as conn:\n with conn.cursor() as cursor:\n conditions = []\n params: List[Any] = []\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(\n \"(title LIKE %s OR description LIKE %s OR content LIKE %s)\"\n )\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(\n f\"SELECT COUNT(*) as count FROM dokumentation WHERE {where_clause}\",\n params,\n )\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 SimpleDbConnection.get_connection(Config) 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,\n 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\n if isinstance(doc.status, DocStatus)\n else d... [TRUNCATED-e6e81cf9864f22cf]",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 52,
"newStart": 1,
"newLines": 51,
"lines": [
"-\"\"\"Repository für Dokumentation CRUD-Operationen\"\"\"",
"-import sys",
"+\"\"\"Repository für Dokumentation CRUD-Operationen.\"\"\"",
"+",
" from datetime import datetime",
"-from typing import Optional, List, Dict, Any",
"+from typing import Any, Dict, List, Optional",
" ",
"-sys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_docs\")",
" from config import Config",
"-from domain.dokumentation import Dokumentation, DocStatus",
"-from infrastructure.db_connection import DatabaseConnection",
"+from domain.dokumentation import DocStatus, Dokumentation",
"+from shared.infrastructure import SimpleDbConnection",
" ",
" ",
" class DocsRepository:",
"- \"\"\"Repository für ki_system.dokumentation Tabelle\"\"\"",
"+ \"\"\"Repository für ki_system.dokumentation Tabelle.\"\"\"",
" ",
" def find_by_id(self, doc_id: int) -> Optional[Dokumentation]:",
"- \"\"\"Findet Dokument nach ID\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ \"\"\"Findet Dokument nach ID.\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" cursor.execute(",
"- \"SELECT * FROM dokumentation WHERE id = %s\",",
"- (doc_id,)",
"+ \"SELECT * FROM dokumentation WHERE id = %s\", (doc_id,)",
" )",
" row = cursor.fetchone()",
" return self._row_to_doc(row) if row else None",
" ",
" def find_by_path(self, path: str) -> Optional[Dokumentation]:",
"- \"\"\"Findet Dokument nach Pfad\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ \"\"\"Findet Dokument nach Pfad.\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" cursor.execute(",
"- \"SELECT * FROM dokumentation WHERE path = %s\",",
"- (path,)",
"+ \"SELECT * FROM dokumentation WHERE path = %s\", (path,)",
" )",
" row = cursor.fetchone()",
" return self._row_to_doc(row) if row else None",
" ",
"- def find_by_slug(self, slug: str, parent_id: Optional[int] = None) -> Optional[Dokumentation]:",
"- \"\"\"Findet Dokument nach Slug (optional mit Parent)\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ def find_by_slug(",
"+ self, slug: str, parent_id: Optional[int] = None",
"+ ) -> Optional[Dokumentation]:",
"+ \"\"\"Findet Dokument nach Slug (optional mit Parent).\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" if parent_id is not None:",
" cursor.execute(",
" \"SELECT * FROM dokumentation WHERE slug = %s AND parent_id = %s\",",
"- (slug, parent_id)",
"+ (slug, parent_id),",
" )",
" else:",
" cursor.execute(",
" \"SELECT * FROM dokumentation WHERE slug = %s AND parent_id IS NULL\",",
"- (slug,)",
"+ (slug,),",
" )",
" row = cursor.fetchone()",
" return self._row_to_doc(row) if row else None"
]
},
{
"oldStart": 57,
"oldLines": 13,
"newStart": 56,
"newLines": 13,
"lines": [
" parent_id: Optional[int] = None,",
" search: Optional[str] = None,",
" limit: int = 20,",
"- offset: int = 0",
"+ offset: int = 0,",
" ) -> List[Dokumentation]:",
"- \"\"\"Findet alle Dokumente mit optionalen Filtern\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ \"\"\"Findet alle Dokumente mit optionalen Filtern.\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" conditions = []",
"- params = []",
"+ params: List[Any] = []",
" ",
" if status:",
" conditions.append(\"status = %s\")"
]
},
{
"oldStart": 74,
"oldLines": 7,
"newStart": 73,
"newLines": 9,
"lines": [
" params.append(parent_id)",
" ",
" if search:",
"- conditions.append(\"(title LIKE %s OR description LIKE %s OR content LIKE %s)\")",
"+ conditions.append(",
"+ \"(title LIKE %s OR description LIKE %s OR content LIKE %s)\"",
"+ )",
" search_term = f\"%{search}%\"",
" params.extend([search_term, search_term, search_term])",
" "
]
},
{
"oldStart": 96,
"oldLines": 21,
"newStart": 97,
"newLines": 21,
"lines": [
" return [self._row_to_doc(row) for row in rows]",
" ",
" def find_children(self, parent_id: int) -> List[Dokumentation]:",
"- \"\"\"Findet alle direkten Kinder eines Dokuments\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ \"\"\"Findet alle direkten Kinder eines Dokuments.\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" cursor.execute(",
" \"\"\"SELECT * FROM dokumentation",
" WHERE parent_id = %s",
" ORDER BY sort_order, title\"\"\",",
"- (parent_id,)",
"+ (parent_id,),",
" )",
" rows = cursor.fetchall()",
" return [self._row_to_doc(row) for row in rows]",
" ",
" def find_root_documents(self) -> List[Dokumentation]:",
"- \"\"\"Findet alle Root-Dokumente (ohne Parent)\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ \"\"\"Findet alle Root-Dokumente (ohne Parent).\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" cursor.execute(",
" \"\"\"SELECT * FROM dokumentation"
]
},
{
"oldStart": 121,
"oldLines": 8,
"newStart": 122,
"newLines": 8,
"lines": [
" return [self._row_to_doc(row) for row in rows]",
" ",
" def get_hierarchy(self) -> List[Dict[str, Any]]:",
"- \"\"\"Gibt kompletten Dokumentationsbaum zurück\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ \"\"\"Gibt kompletten Dokumentationsbaum zurück.\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" cursor.execute(",
" \"\"\"SELECT id, parent_id, slug, path, title, status, sort_order, depth"
]
},
{
"oldStart": 145,
"oldLines": 24,
"newStart": 146,
"newLines": 27,
"lines": [
" return roots",
" ",
" def get_breadcrumb(self, doc_id: int) -> List[Dict[str, Any]]:",
"- \"\"\"Gibt Breadcrumb-Pfad für ein Dokument zurück\"\"\"",
"- breadcrumb = []",
"- current_id = doc_id",
"+ \"\"\"Gibt Breadcrumb-Pfad für ein Dokument zurück.\"\"\"",
"+ breadcrumb: List[Dict[str, Any]] = []",
"+ current_id: Optional[int] = doc_id",
" ",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" while current_id is not None:",
" cursor.execute(",
" \"SELECT id, parent_id, path, title FROM dokumentation WHERE id = %s\",",
"- (current_id,)",
"+ (current_id,),",
" )",
" row = cursor.fetchone()",
" if row:",
"- breadcrumb.insert(0, {",
"- \"id\": row[\"id\"],",
"- \"path\": row[\"path\"],",
"- \"title\": row[\"title\"]",
"- })",
"+ breadcrumb.insert(",
"+ 0,",
"+ {",
"+ \"id\": row[\"id\"],",
"+ \"path\": row[\"path\"],",
"+ \"title\": row[\"title\"],",
"+ },",
"+ )",
" current_id = row[\"parent_id\"]",
" else:",
" break"
]
},
{
"oldStart": 170,
"oldLines": 26,
"newStart": 174,
"newLines": 26,
"lines": [
" return breadcrumb",
" ",
" def get_siblings(self, doc_id: int) -> List[Dokumentation]:",
"- \"\"\"Findet Geschwister-Dokumente (gleicher Parent)\"\"\"",
"+ \"\"\"Findet Geschwister-Dokumente (gleicher Parent).\"\"\"",
" doc = self.find_by_id(doc_id)",
" if not doc:",
" return []",
" ",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" if doc.parent_id is not None:",
" cursor.execute(",
" \"\"\"SELECT * FROM dokumentation",
" WHERE parent_id = %s AND id != %s",
" ORDER BY sort_order, title\"\"\",",
"- (doc.parent_id, doc_id)",
"+ (doc.parent_id, doc_id),",
" )",
" else:",
" cursor.execute(",
" \"\"\"SELECT * FROM dokumentation",
" WHERE parent_id IS NULL AND id != %s",
" ORDER BY sort_order, title\"\"\",",
"- (doc_id,)",
"+ (doc_id,),",
" )",
" rows = cursor.fetchall()",
" return [self._row_to_doc(row) for row in rows]"
]
},
{
"oldStart": 198,
"oldLines": 13,
"newStart": 202,
"newLines": 13,
"lines": [
" self,",
" status: Optional[str] = None,",
" parent_id: Optional[int] = None,",
"- search: Optional[str] = None",
"+ search: Optional[str] = None,",
" ) -> int:",
"- \"\"\"Zählt Dokumente mit optionalen Filtern\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ \"\"\"Zählt Dokumente mit optionalen Filtern.\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" conditions = []",
"- params = []",
"+ params: List[Any] = []",
" ",
" if status:",
" conditions.append(\"status = %s\")"
]
},
{
"oldStart": 215,
"oldLines": 24,
"newStart": 219,
"newLines": 30,
"lines": [
" params.append(parent_id)",
" ",
" if search:",
"- conditions.append(\"(title LIKE %s OR description LIKE %s OR content LIKE %s)\")",
"+ conditions.append(",
"+ \"(title LIKE %s OR description LIKE %s OR content LIKE %s)\"",
"+ )",
" search_term = f\"%{search}%\"",
" params.extend([search_term, search_term, search_term])",
" ",
" where_clause = \" AND \".join(conditions) if conditions else \"1=1\"",
" ",
"- cursor.execute(f\"SELECT COUNT(*) as count FROM dokumentation WHERE {where_clause}\", params)",
"+ cursor.execute(",
"+ f\"SELECT COUNT(*) as count FROM dokumentation WHERE {where_clause}\",",
"+ params,",
"+ )",
" result = cursor.fetchone()",
" return result[\"count\"] if result else 0",
" ",
" def create(self, doc: Dokumentation) -> int:",
"- \"\"\"Erstellt neues Dokument, gibt ID zurück\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ \"\"\"Erstellt neues Dokument, gibt ID zurück.\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" now = datetime.now()",
" cursor.execute(",
" \"\"\"INSERT INTO dokumentation",
"- (parent_id, slug, path, title, description, content, status, sort_order, depth, created_at, updated_at)",
"+ (parent_id, slug, path, title, description, content, status,",
"+ sort_order, depth, created_at, updated_at)",
" VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)\"\"\",",
" (",
" doc.parent_id,"
]
},
{
"oldStart": 241,
"oldLines": 21,
"newStart": 251,
"newLines": 33,
"lines": [
" doc.title,",
" doc.description,",
" doc.content,",
"- doc.status.value if isinstance(doc.status, DocStatus) else doc.status,",
"+ doc.status.value",
"+ if isinstance(doc.status, DocStatus)",
"+ else doc.status,",
" doc.sort_order,",
" doc.depth,",
" now,",
"- now",
"- )",
"+ now,",
"+ ),",
" )",
"- return cursor.lastrowid",
"+ return cursor.lastrowid or 0",
" ",
" def update(self, doc_id: int, updates: Dict[str, Any]) -> bool:",
"- \"\"\"Aktualisiert Dokument-Felder\"\"\"",
"+ \"\"\"Aktualisiert Dokument-Felder.\"\"\"",
" if not updates:",
" return False",
" ",
"- allowed_fields = {\"title\", \"description\", \"content\", \"status\", \"slug\", \"path\", \"sort_order\", \"parent_id\", \"depth\"}",
"+ allowed_fields = {",
"+ \"title\",",
"+ \"description\",",
"+ \"content\",",
"+ \"status\",",
"+ \"slug\",",
"+ \"path\",",
"+ \"sort_order\",",
"+ \"parent_id\",",
"+ \"depth\",",
"+ }",
" filtered_updates = {k: v for k, v in updates.items() if k in allowed_fields}",
" ",
" if not filtered_updates:"
]
},
{
"oldStart": 264,
"oldLines": 7,
"newStart": 286,
"newLines": 7,
"lines": [
" # updated_at automatisch setzen",
" filtered_updates[\"updated_at\"] = datetime.now()",
" ",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" set_clause = \", \".join([f\"{k} = %s\" for k in filtered_updates.keys()])",
" values = list(filtered_updates.values())"
]
},
{
"oldStart": 272,
"oldLines": 33,
"newStart": 294,
"newLines": 38,
"lines": [
" ",
" cursor.execute(",
" f\"UPDATE dokumentation SET {set_clause} WHERE id = %s\",",
"- values",
"+ values,",
" )",
" return cursor.rowcount > 0",
" ",
" def delete(self, doc_id: int) -> bool:",
"- \"\"\"Löscht ein Dokument (keine Kaskade!)\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ \"\"\"Löscht ein Dokument (keine Kaskade!).\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" # Prüfen ob Kinder existieren",
" cursor.execute(",
" \"SELECT COUNT(*) as count FROM dokumentation WHERE parent_id = %s\",",
"- (doc_id,)",
"+ (doc_id,),",
" )",
" result = cursor.fetchone()",
" if result and result[\"count\"] > 0:",
"- raise ValueError(f\"Dokument hat {result['count']} Kind-Dokumente. Bitte zuerst diese löschen.\")",
"+ raise ValueError(",
"+ f\"Dokument hat {result['count']} Kind-Dokumente. \"",
"+ \"Bitte zuerst diese löschen.\"",
"+ )",
" ",
" cursor.execute(\"DELETE FROM dokumentation WHERE id = %s\", (doc_id,))",
" return cursor.rowcount > 0",
" ",
"- def move(self, doc_id: int, new_parent_id: Optional[int], new_sort_order: Optional[int] = None) -> bool:",
"- \"\"\"Verschiebt Dokument zu neuem Parent\"\"\"",
"+ def move(",
"+ self, doc_id: int, new_parent_id: Optional[int], new_sort_order: Optional[int] = None",
"+ ) -> bool:",
"+ \"\"\"Verschiebt Dokument zu neuem Parent.\"\"\"",
" doc = self.find_by_id(doc_id)",
" if not doc:",
" return False",
" ",
"- updates = {\"parent_id\": new_parent_id}",
"+ updates: Dict[str, Any] = {\"parent_id\": new_parent_id}",
" ",
" # Neue Tiefe berechnen",
" if new_parent_id is not None:"
]
},
{
"oldStart": 317,
"oldLines": 11,
"newStart": 344,
"newLines": 12,
"lines": [
" return self.update(doc_id, updates)",
" ",
" def get_statistics(self) -> Dict[str, Any]:",
"- \"\"\"Gibt Statistiken über alle Dokumente zurück\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ \"\"\"Gibt Statistiken über alle Dokumente zurück.\"\"\"",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" # Gesamt und nach Status",
"- cursor.execute(\"\"\"",
"+ cursor.execute(",
"+ \"\"\"",
" SELECT",
" COUNT(*) as total,",
" SUM(CASE WHEN status = 'draft' THEN 1 ELSE 0 END) as draft,"
]
},
{
"oldStart": 329,
"oldLines": 11,
"newStart": 357,
"newLines": 14,
"lines": [
" SUM(CASE WHEN status = 'archived' THEN 1 ELSE 0 END) as archived,",
" MAX(depth) as max_depth",
" FROM dokumentation",
"- \"\"\")",
"+ \"\"\"",
"+ )",
" stats = cursor.fetchone()",
" ",
" # Root-Dokumente zählen",
"- cursor.execute(\"SELECT COUNT(*) as count FROM dokumentation WHERE parent_id IS NULL\")",
"+ cursor.execute(",
"+ \"SELECT COUNT(*) as count FROM dokumentation WHERE parent_id IS NULL\"",
"+ )",
" root_count = cursor.fetchone()",
" ",
" return {"
]
},
{
"oldStart": 341,
"oldLines": 14,
"newStart": 372,
"newLines": 14,
"lines": [
" \"by_status\": {",
" \"draft\": stats[\"draft\"] or 0,",
" \"published\": stats[\"published\"] or 0,",
"- \"archived\": stats[\"archived\"] or 0",
"+ \"archived\": stats[\"archived\"] or 0,",
" },",
" \"root_documents\": root_count[\"count\"] if root_count else 0,",
"- \"max_depth\": stats[\"max_depth\"] or 0",
"+ \"max_depth\": stats[\"max_depth\"] or 0,",
" }",
" ",
" def _row_to_doc(self, row: Dict[str, Any]) -> Dokumentation:",
"- \"\"\"Konvertiert DB-Row zu Dokumentation-Entity\"\"\"",
"+ \"\"\"Konvertiert DB-Row zu Dokumentation-Entity.\"\"\"",
" status = row.get(\"status\", \"draft\")",
" if isinstance(status, str):",
" try:"
]
},
{
"oldStart": 368,
"oldLines": 16,
"newStart": 399,
"newLines": 16,
"lines": [
" sort_order=row.get(\"sort_order\", 0),",
" depth=row.get(\"depth\", 0),",
" created_at=row.get(\"created_at\", datetime.now()),",
"- updated_at=row.get(\"updated_at\", datetime.now())",
"+ updated_at=row.get(\"updated_at\", datetime.now()),",
" )",
" ",
" ",
" # Singleton-Instanz",
"-_repository_instance = None",
"+_repository_instance: Optional[DocsRepository] = None",
" ",
" ",
" def get_repository() -> DocsRepository:",
"- \"\"\"Gibt Singleton-Repository zurück\"\"\"",
"+ \"\"\"Gibt Singleton-Repository zurück.\"\"\"",
" global _repository_instance",
" if _repository_instance is None:",
" _repository_instance = DocsRepository()"
]
}
],
"originalFile": "\"\"\"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]"
}
}