{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/js_error_repository.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterable\nfrom typing import Any\n\nfrom src.domain.entities.js_error import JsError\nfrom src.domain.ports.js_error_repository_port import JsErrorRepositoryPort\nfrom src.domain.value_objects.enums import JsErrorSeverity\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_INSERT = (\n \"INSERT INTO js_errors (page_id, severity, message, source, line_no, col_no)\"\n \" VALUES (%(page_id)s, %(severity)s, %(message)s, %(source)s, %(line_no)s, %(col_no)s)\"\n)\n_SELECT_BY_PAGE = \"SELECT * FROM js_errors WHERE page_id=%s ORDER BY id\"\n\n\nclass JsErrorRepository(JsErrorRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, errors: Iterable[JsError]) -> None:\n rows = [_to_row(e) for e in errors]\n if not rows:\n return\n with self._connections.session() as conn:\n conn.cursor().executemany(_INSERT, rows)\n\n def list_by_page(self, page_id: int) -> list[JsError]:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_PAGE, (page_id,))\n return [_to_entity(r) for r in cur.fetchall()]\n\n\ndef _to_row(e: JsError) -> dict[str, Any]:\n return {\n \"page_id\": e.page_id, \"severity\": e.severity.value, \"message\": e.message,\n \"source\": e.source, \"line_no\": e.line_no, \"col_no\": e.col_no,\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> JsError:\n return JsError(\n id=row[\"id\"], page_id=row[\"page_id\"],\n severity=JsErrorSeverity(row[\"severity\"]), message=row[\"message\"],\n source=row.get(\"source\"), line_no=row.get(\"line_no\"), col_no=row.get(\"col_no\"),\n )\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/js_error_repository.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterable\nfrom typing import Any\n\nfrom src.domain.entities.js_error import JsError\nfrom src.domain.ports.js_error_repository_port import JsErrorRepositoryPort\nfrom src.domain.value_objects.enums import JsErrorSeverity\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_INSERT = (\n \"INSERT INTO js_errors (page_id, severity, message, source, line_no, col_no)\"\n \" VALUES (%(page_id)s, %(severity)s, %(message)s, %(source)s, %(line_no)s, %(col_no)s)\"\n)\n_SELECT_BY_PAGE = \"SELECT * FROM js_errors WHERE page_id=%s ORDER BY id\"\n\n\nclass JsErrorRepository(JsErrorRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, errors: Iterable[JsError]) -> None:\n rows = [_to_row(e) for e in errors]\n if not rows:\n return\n with self._connections.session() as conn:\n conn.cursor().executemany(_INSERT, rows)\n\n def list_by_page(self, page_id: int) -> list[JsError]:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_PAGE, (page_id,))\n return [_to_entity(r) for r in cur.fetchall()]\n\n\ndef _to_row(e: JsError) -> dict[str, Any]:\n return {\n \"page_id\": e.page_id, \"severity\": e.severity.value, \"message\": e.message,\n \"source\": e.source, \"line_no\": e.line_no, \"col_no\": e.col_no,\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> JsError:\n return JsError(\n id=row[\"id\"], page_id=row[\"page_id\"],\n severity=JsErrorSeverity(row[\"severity\"]), message=row[\"message\"],\n source=row.get(\"source\"), line_no=row.get(\"line_no\"), col_no=row.get(\"col_no\"),\n )\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}