{
"tool_response": {
"filePath": "\/var\/www\/mcp-servers\/mcp_contracts\/infrastructure\/contract_repository.py",
"oldString": "\"\"\"Contract Repository - Datenbankzugriff für Contracts\"\"\"\nimport sys\nimport uuid\nimport json\nfrom datetime import datetime\nfrom typing import Optional, List, Dict, Any\n\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_contracts\")\nfrom domain.contracts import Contract, ContractHistory, ContractValidation, ContractStatus\nfrom infrastructure.db_connection import DatabaseConnection",
"newString": "\"\"\"Contract Repository - Datenbankzugriff für Contracts.\"\"\"\n\nimport json\nimport uuid\nfrom datetime import datetime\nfrom typing import Any, Dict, List, Optional\n\nfrom config import Config\nfrom domain.contracts import Contract, ContractHistory, ContractStatus, ContractValidation\nfrom shared.infrastructure import SimpleDbConnection",
"originalFile": "\"\"\"Contract Repository - Datenbankzugriff für Contracts\"\"\"\nimport sys\nimport uuid\nimport json\nfrom datetime import datetime\nfrom typing import Optional, List, Dict, Any\n\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_contracts\")\nfrom domain.contracts import Contract, ContractHistory, ContractValidation, ContractStatus\nfrom infrastructure.db_connection import DatabaseConnection\n\n\nclass ContractRepository:\n \"\"\"Repository für Contract-CRUD-Operationen\"\"\"\n\n # ==================== CONTRACT CRUD ====================\n\n def find_by_id(self, contract_id: int) -> Optional[Contract]:\n \"\"\"Findet Contract nach ID\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n cursor.execute(\"SELECT * FROM contracts WHERE id = %s\", (contract_id,))\n row = cursor.fetchone()\n return self._row_to_contract(row) if row else None\n\n def find_by_name(self, name: str, version: Optional[str] = None) -> Optional[Contract]:\n \"\"\"Findet Contract nach Name (optional: spezifische Version)\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n if version:\n cursor.execute(\n \"SELECT * FROM contracts WHERE name = %s AND version = %s\",\n (name, version)\n )\n else:\n # Neueste aktive Version\n cursor.execute(\n \"\"\"SELECT * FROM contracts\n WHERE name = %s AND status = 'active'\n ORDER BY version DESC LIMIT 1\"\"\",\n (name,)\n )\n row = cursor.fetchone()\n return self._row_to_contract(row) if row else None\n\n def find_all(\n self,\n status: Optional[str] = None,\n search: Optional[str] = None,\n limit: int = 50,\n offset: int = 0,\n ) -> List[Contract]:\n \"\"\"Findet Contracts mit Filtern\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n sql = \"SELECT * FROM contracts WHERE 1=1\"\n params = []\n\n if status:\n sql += \" AND status = %s\"\n params.append(status)\n\n if search:\n sql += \" AND (name LIKE %s OR scope_description LIKE %s)\"\n params.extend([f\"%{search}%\", f\"%{search}%\"])\n\n sql += \" ORDER BY name ASC, version DESC LIMIT %s OFFSET %s\"\n params.extend([limit, offset])\n\n cursor.execute(sql, params)\n rows = cursor.fetchall()\n return [self._row_to_contract(row) for row in rows]\n\n def count(self, status: Optional[str] = None, search: Optional[str] = None) -> int:\n \"\"\"Zählt Contracts mit Filtern\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n sql = \"SELECT COUNT(*) as cnt FROM contracts WHERE 1=1\"\n params = []\n\n if status:\n sql += \" AND status = %s\"\n params.append(status)\n\n if search:\n sql += \" AND (name LIKE %s OR scope_description LIKE %s)\"\n params.extend([f\"%{search}%\", f\"%{search}%\"])\n\n cursor.execute(sql, params)\n row = cursor.fetchone()\n return row[\"cnt\"] if row else 0\n\n def create(self, contract: Contract) -> int:\n \"\"\"Erstellt neuen Contract, gibt ID zurück\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n contract.uuid = str(uuid.uuid4())\n contract.created_at = datetime.now()\n contract.updated_at = datetime.now()\n\n sql = \"\"\"\n INSERT INTO contracts\n (uuid, name, version, status, yaml_content, scope_description,\n created_at, created_by, updated_at)\n VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)\n \"\"\"\n status_value = contract.status.value if isinstance(contract.status, ContractStatus) else contract.status\n cursor.execute(sql, (\n contract.uuid,\n contract.name,\n contract.version,\n status_value,\n contract.yaml_content,\n contract.scope_description,\n contract.created_at,\n contract.created_by,\n contract.updated_at,\n ))\n conn.commit()\n return cursor.lastrowid\n\n def update(self, contract_id: int, updates: Dict[str, Any]) -> bool:\n \"\"\"Aktualisiert Contract-Felder\"\"\"\n if not updates:\n return False\n\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n set_parts = []\n params = []\n\n for key, value in updates.items():\n if key in [\"name\", \"version\", \"status\", \"yaml_content\", \"scope_description\"]:\n set_parts.append(f\"{key} = %s\")\n if key == \"status\" and isinstance(value, ContractStatus):\n params.append(value.value)\n else:\n params.append(value)\n\n if not set_parts:\n return False\n\n set_parts.append(\"updated_at = %s\")\n params.append(datetime.now())\n params.append(contract_id)\n\n sql = f\"UPDATE contracts SET {', '.join(set_parts)} WHERE id = %s\"\n cursor.execute(sql, params)\n conn.commit()\n return cursor.rowcount > 0\n\n def delete(self, contract_id: int) -> bool:\n \"\"\"Löscht Contract (cascade löscht auch history und validations)\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n cursor.execute(\"DELETE FROM contracts WHERE id = %s\", (contract_id,))\n conn.commit()\n return cursor.rowcount > 0\n\n def deprecate(self, contract_id: int) -> bool:\n \"\"\"Markiert Contract als deprecated\"\"\"\n return self.update(contract_id, {\"status\": \"deprecated\"})\n\n # ==================== VERSIONING ====================\n\n def create_new_version(\n self,\n contract_id: int,\n new_yaml: str,\n new_version: str,\n change_description: str,\n changed_by: str = \"mcp-contracts\"\n ) -> int:\n \"\"\"Erstellt neue Version eines Contracts\"\"\"\n old_contract = self.find_by_id(contract_id)\n if not old_contract:\n raise ValueError(f\"Contract {contract_id} not found\")\n\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n # Historie speichern\n cursor.execute(\"\"\"\n INSERT INTO contract_history\n (contract_id, previous_yaml, previous_version, change_description, changed_by)\n VALUES (%s, %s, %s, %s, %s)\n \"\"\", (\n contract_id,\n old_contract.yaml_content,\n old_contract.version,\n change_description,\n changed_by,\n ))\n\n # Contract aktualisieren\n cursor.execute(\"\"\"\n UPDATE contracts\n SET yaml_content = %s, version = %s, updated_at = %s\n WHERE id = %s\n \"\"\", (new_yaml, new_version, datetime.now(), contract_id))\n\n conn.commit()\n return contract_id\n\n def get_versions(self, contract_id: int) -> List[Dict[str, Any]]:\n \"\"\"Holt alle Versionen eines Contracts\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n # Aktuelle Version\n cursor.execute(\n \"SELECT version, updated_at, 'current' as type FROM contracts WHERE id = %s\",\n (contract_id,)\n )\n current = cursor.fetchone()\n\n # Historische Versionen\n cursor.execute(\"\"\"\n SELECT previous_version as version, changed_at as updated_at,\n 'history' as type, change_description, changed_by\n FROM contract_history\n WHERE contract_id = %s\n ORDER BY changed_at DESC\n \"\"\", (contract_id,))\n history = cursor.fetchall()\n\n versions = []\n if current:\n versions.append({\n \"version\": current[\"version\"],\n \"updated_at\": current[\"updated_at\"].isoformat() if current[\"updated_at\"] else None,\n \"type\": \"current\",\n })\n for h in history:\n versions.append({\n \"version\": h[\"version\"],\n \"updated_at\": h[\"updated_at\"].isoformat() if h[\"updated_at\"] else None,\n \"type\": \"history\",\n \"change_description\": h[\"change_description\"],\n \"changed_by\": h[\"changed_by\"],\n })\n return versions\n\n # ==================== HISTORY ====================\n\n def get_history(self, contract_id: int) -> List[ContractHistory]:\n \"\"\"Holt Änderungshistorie eines Contracts\"\"\"\n with DatabaseConnection.get_connection() as conn:\n with conn.cursor() as cursor:\n cursor.execute(\"\"\"\n SE... [TRUNCATED-16f6b2ac26425a69]",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 13,
"newStart": 1,
"newLines": 13,
"lines": [
"-\"\"\"Contract Repository - Datenbankzugriff für Contracts\"\"\"",
"-import sys",
"-import uuid",
"+\"\"\"Contract Repository - Datenbankzugriff für Contracts.\"\"\"",
"+",
" import json",
"+import uuid",
" 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_contracts\")",
"-from domain.contracts import Contract, ContractHistory, ContractValidation, ContractStatus",
"-from infrastructure.db_connection import DatabaseConnection",
"+from config import Config",
"+from domain.contracts import Contract, ContractHistory, ContractStatus, ContractValidation",
"+from shared.infrastructure import SimpleDbConnection",
" ",
" ",
" class ContractRepository:"
]
}
],
"userModified": false,
"replaceAll": false
}
}