{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/hreflang_repository.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterable\n\nfrom src.domain.ports.hreflang_repository_port import HreflangRepositoryPort\nfrom src.domain.value_objects.hreflang import HreflangEntry\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_INSERT = \"INSERT INTO hreflang (page_id, lang, href) VALUES (%s, %s, %s)\"\n_SELECT_BY_PAGE = \"SELECT lang, href FROM hreflang WHERE page_id=%s ORDER BY id\"\n\n\nclass HreflangRepository(HreflangRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, page_id: int, entries: Iterable[HreflangEntry]) -> None:\n rows = [(page_id, e.lang, e.href) for e in entries]\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[HreflangEntry]:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_PAGE, (page_id,))\n return [HreflangEntry(lang=r[\"lang\"], href=r[\"href\"]) for r in cur.fetchall()]\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/hreflang_repository.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterable\n\nfrom src.domain.ports.hreflang_repository_port import HreflangRepositoryPort\nfrom src.domain.value_objects.hreflang import HreflangEntry\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_INSERT = \"INSERT INTO hreflang (page_id, lang, href) VALUES (%s, %s, %s)\"\n_SELECT_BY_PAGE = \"SELECT lang, href FROM hreflang WHERE page_id=%s ORDER BY id\"\n\n\nclass HreflangRepository(HreflangRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, page_id: int, entries: Iterable[HreflangEntry]) -> None:\n rows = [(page_id, e.lang, e.href) for e in entries]\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[HreflangEntry]:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_PAGE, (page_id,))\n return [HreflangEntry(lang=r[\"lang\"], href=r[\"href\"]) for r in cur.fetchall()]\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}