{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/connection.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterator\nfrom contextlib import contextmanager\nfrom typing import Any\n\nimport mysql.connector\n\nfrom src.infrastructure.config.db_config import DbSettings\n\n\nclass MariaDbConnectionFactory:\n def __init__(self, settings: DbSettings) -> None:\n self._settings = settings\n\n def connect(self) -> Any: # noqa: ANN401 — mysql-connector types are dynamic\n return mysql.connector.connect(\n host=self._settings.host,\n user=self._settings.user,\n password=self._settings.password,\n database=self._settings.database,\n autocommit=False,\n )\n\n @contextmanager\n def session(self) -> Iterator[Any]: # noqa: ANN401 — mysql-connector types\n conn = self.connect()\n try:\n yield conn\n conn.commit()\n except Exception:\n conn.rollback()\n raise\n finally:\n conn.close()\n"
}
}
{
"tool_response": {
"type": "update",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/connection.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterator\nfrom contextlib import contextmanager\nfrom typing import Any\n\nimport mysql.connector\n\nfrom src.infrastructure.config.db_config import DbSettings\n\n\nclass MariaDbConnectionFactory:\n def __init__(self, settings: DbSettings) -> None:\n self._settings = settings\n\n def connect(self) -> Any: # noqa: ANN401 — mysql-connector types are dynamic\n return mysql.connector.connect(\n host=self._settings.host,\n user=self._settings.user,\n password=self._settings.password,\n database=self._settings.database,\n autocommit=False,\n )\n\n @contextmanager\n def session(self) -> Iterator[Any]: # noqa: ANN401 — mysql-connector types\n conn = self.connect()\n try:\n yield conn\n conn.commit()\n except Exception:\n conn.rollback()\n raise\n finally:\n conn.close()\n",
"structuredPatch": [
{
"oldStart": 13,
"oldLines": 7,
"newStart": 13,
"newLines": 7,
"lines": [
" def __init__(self, settings: DbSettings) -> None:",
" self._settings = settings",
" ",
"- def connect(self) -> Any: # mysql.connector types are dynamic",
"+ def connect(self) -> Any: # noqa: ANN401 — mysql-connector types are dynamic",
" return mysql.connector.connect(",
" host=self._settings.host,",
" user=self._settings.user,"
]
},
{
"oldStart": 23,
"oldLines": 7,
"newStart": 23,
"newLines": 7,
"lines": [
" )",
" ",
" @contextmanager",
"- def session(self) -> Iterator[Any]:",
"+ def session(self) -> Iterator[Any]: # noqa: ANN401 — mysql-connector types",
" conn = self.connect()",
" try:",
" yield conn"
]
}
],
"originalFile": "from __future__ import annotations\n\nfrom collections.abc import Iterator\nfrom contextlib import contextmanager\nfrom typing import Any\n\nimport mysql.connector\n\nfrom src.infrastructure.config.db_config import DbSettings\n\n\nclass MariaDbConnectionFactory:\n def __init__(self, settings: DbSettings) -> None:\n self._settings = settings\n\n def connect(self) -> Any: # mysql.connector types are dynamic\n return mysql.connector.connect(\n host=self._settings.host,\n user=self._settings.user,\n password=self._settings.password,\n database=self._settings.database,\n autocommit=False,\n )\n\n @contextmanager\n def session(self) -> Iterator[Any]:\n conn = self.connect()\n try:\n yield conn\n conn.commit()\n except Exception:\n conn.rollback()\n raise\n finally:\n conn.close()\n",
"userModified": false
}
}