{
"tool_response": {
"filePath": "\/var\/www\/mcp-servers\/mcp_contracts\/infrastructure\/contract_repository.py",
"oldString": "DatabaseConnection.get_connection()",
"newString": "SimpleDbConnection.get_connection(Config)",
"originalFile": "\"\"\"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\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 SELECT * FROM contract_history\n ... [TRUNCATED-dbb40aeec68f8835]",
"structuredPatch": [
{
"oldStart": 17,
"oldLines": 7,
"newStart": 17,
"newLines": 7,
"lines": [
" ",
" def find_by_id(self, contract_id: int) -> Optional[Contract]:",
" \"\"\"Findet Contract nach ID\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" cursor.execute(\"SELECT * FROM contracts WHERE id = %s\", (contract_id,))",
" row = cursor.fetchone()"
]
},
{
"oldStart": 25,
"oldLines": 7,
"newStart": 25,
"newLines": 7,
"lines": [
" ",
" def find_by_name(self, name: str, version: Optional[str] = None) -> Optional[Contract]:",
" \"\"\"Findet Contract nach Name (optional: spezifische Version)\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" if version:",
" cursor.execute("
]
},
{
"oldStart": 51,
"oldLines": 7,
"newStart": 51,
"newLines": 7,
"lines": [
" offset: int = 0,",
" ) -> List[Contract]:",
" \"\"\"Findet Contracts mit Filtern\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" sql = \"SELECT * FROM contracts WHERE 1=1\"",
" params = []"
]
},
{
"oldStart": 73,
"oldLines": 7,
"newStart": 73,
"newLines": 7,
"lines": [
" ",
" def count(self, status: Optional[str] = None, search: Optional[str] = None) -> int:",
" \"\"\"Zählt Contracts mit Filtern\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" sql = \"SELECT COUNT(*) as cnt FROM contracts WHERE 1=1\"",
" params = []"
]
},
{
"oldStart": 92,
"oldLines": 7,
"newStart": 92,
"newLines": 7,
"lines": [
" ",
" def create(self, contract: Contract) -> int:",
" \"\"\"Erstellt neuen Contract, gibt ID zurück\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" contract.uuid = str(uuid.uuid4())",
" contract.created_at = datetime.now()"
]
},
{
"oldStart": 124,
"oldLines": 7,
"newStart": 124,
"newLines": 7,
"lines": [
" if not updates:",
" return False",
" ",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" set_parts = []",
" params = []"
]
},
{
"oldStart": 151,
"oldLines": 7,
"newStart": 151,
"newLines": 7,
"lines": [
" ",
" def delete(self, contract_id: int) -> bool:",
" \"\"\"Löscht Contract (cascade löscht auch history und validations)\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" cursor.execute(\"DELETE FROM contracts WHERE id = %s\", (contract_id,))",
" conn.commit()"
]
},
{
"oldStart": 176,
"oldLines": 7,
"newStart": 176,
"newLines": 7,
"lines": [
" if not old_contract:",
" raise ValueError(f\"Contract {contract_id} not found\")",
" ",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" # Historie speichern",
" cursor.execute(\"\"\""
]
},
{
"oldStart": 203,
"oldLines": 7,
"newStart": 203,
"newLines": 7,
"lines": [
" ",
" def get_versions(self, contract_id: int) -> List[Dict[str, Any]]:",
" \"\"\"Holt alle Versionen eines Contracts\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" # Aktuelle Version",
" cursor.execute("
]
},
{
"oldStart": 243,
"oldLines": 7,
"newStart": 243,
"newLines": 7,
"lines": [
" ",
" def get_history(self, contract_id: int) -> List[ContractHistory]:",
" \"\"\"Holt Änderungshistorie eines Contracts\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" cursor.execute(\"\"\"",
" SELECT * FROM contract_history"
]
},
{
"oldStart": 257,
"oldLines": 7,
"newStart": 257,
"newLines": 7,
"lines": [
" ",
" def save_validation(self, validation: ContractValidation) -> int:",
" \"\"\"Speichert Validierungsergebnis\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" violations_json = json.dumps(validation.violations) if validation.violations else None",
" "
]
},
{
"oldStart": 287,
"oldLines": 7,
"newStart": 287,
"newLines": 7,
"lines": [
" limit: int = 10,",
" ) -> List[ContractValidation]:",
" \"\"\"Holt letzte Validierungen eines Contracts\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" cursor.execute(\"\"\"",
" SELECT * FROM contract_validations"
]
},
{
"oldStart": 307,
"oldLines": 7,
"newStart": 307,
"newLines": 7,
"lines": [
" ",
" def get_statistics(self) -> Dict[str, Any]:",
" \"\"\"Holt Contract-Statistiken\"\"\"",
"- with DatabaseConnection.get_connection() as conn:",
"+ with SimpleDbConnection.get_connection(Config) as conn:",
" with conn.cursor() as cursor:",
" # Contracts nach Status",
" cursor.execute(\"\"\""
]
}
],
"userModified": false,
"replaceAll": true
}
}