{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/page_header_repository.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterable, Mapping\n\nfrom src.domain.ports.page_header_repository_port import PageHeaderRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_INSERT = (\n \"INSERT INTO page_headers (page_id, header_name, header_value) VALUES (%s, %s, %s)\"\n)\n_SELECT_BY_PAGE = (\n \"SELECT header_name, header_value FROM page_headers WHERE page_id=%s\"\n)\n\n\nclass PageHeaderRepository(PageHeaderRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, page_id: int, headers: Iterable[tuple[str, str]]) -> None:\n rows = [(page_id, name, value) for name, value in headers]\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) -> Mapping[str, str]:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_PAGE, (page_id,))\n return {r[\"header_name\"]: r[\"header_value\"] for r in cur.fetchall()}\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/page_header_repository.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterable, Mapping\n\nfrom src.domain.ports.page_header_repository_port import PageHeaderRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_INSERT = (\n \"INSERT INTO page_headers (page_id, header_name, header_value) VALUES (%s, %s, %s)\"\n)\n_SELECT_BY_PAGE = (\n \"SELECT header_name, header_value FROM page_headers WHERE page_id=%s\"\n)\n\n\nclass PageHeaderRepository(PageHeaderRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, page_id: int, headers: Iterable[tuple[str, str]]) -> None:\n rows = [(page_id, name, value) for name, value in headers]\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) -> Mapping[str, str]:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_PAGE, (page_id,))\n return {r[\"header_name\"]: r[\"header_value\"] for r in cur.fetchall()}\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}