{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/image_repository.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterable\nfrom typing import Any\n\nfrom src.domain.entities.image import Image\nfrom src.domain.ports.image_repository_port import ImageRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_COLUMNS = (\n \"page_id, src, alt, title, width, height, loading_attr, decoding_attr,\"\n \" http_status, content_length\"\n)\n_PLACEHOLDERS = \", \".join(f\"%({n.strip()})s\" for n in _COLUMNS.split(\",\"))\n_INSERT = f\"INSERT INTO images ({_COLUMNS}) VALUES ({_PLACEHOLDERS})\"\n_SELECT_BY_PAGE = \"SELECT * FROM images WHERE page_id=%s ORDER BY id\"\n\n\nclass ImageRepository(ImageRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, images: Iterable[Image]) -> None:\n rows = [_to_row(i) for i in images]\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[Image]:\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(img: Image) -> dict[str, Any]:\n return {\n \"page_id\": img.page_id, \"src\": img.src, \"alt\": img.alt, \"title\": img.title,\n \"width\": img.width, \"height\": img.height, \"loading_attr\": img.loading_attr,\n \"decoding_attr\": img.decoding_attr, \"http_status\": img.http_status,\n \"content_length\": img.content_length,\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> Image:\n return Image(\n id=row[\"id\"], page_id=row[\"page_id\"], src=row[\"src\"], alt=row.get(\"alt\"),\n title=row.get(\"title\"), width=row.get(\"width\"), height=row.get(\"height\"),\n loading_attr=row.get(\"loading_attr\"), decoding_attr=row.get(\"decoding_attr\"),\n http_status=row.get(\"http_status\"), content_length=row.get(\"content_length\"),\n )\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/image_repository.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterable\nfrom typing import Any\n\nfrom src.domain.entities.image import Image\nfrom src.domain.ports.image_repository_port import ImageRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_COLUMNS = (\n \"page_id, src, alt, title, width, height, loading_attr, decoding_attr,\"\n \" http_status, content_length\"\n)\n_PLACEHOLDERS = \", \".join(f\"%({n.strip()})s\" for n in _COLUMNS.split(\",\"))\n_INSERT = f\"INSERT INTO images ({_COLUMNS}) VALUES ({_PLACEHOLDERS})\"\n_SELECT_BY_PAGE = \"SELECT * FROM images WHERE page_id=%s ORDER BY id\"\n\n\nclass ImageRepository(ImageRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, images: Iterable[Image]) -> None:\n rows = [_to_row(i) for i in images]\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[Image]:\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(img: Image) -> dict[str, Any]:\n return {\n \"page_id\": img.page_id, \"src\": img.src, \"alt\": img.alt, \"title\": img.title,\n \"width\": img.width, \"height\": img.height, \"loading_attr\": img.loading_attr,\n \"decoding_attr\": img.decoding_attr, \"http_status\": img.http_status,\n \"content_length\": img.content_length,\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> Image:\n return Image(\n id=row[\"id\"], page_id=row[\"page_id\"], src=row[\"src\"], alt=row.get(\"alt\"),\n title=row.get(\"title\"), width=row.get(\"width\"), height=row.get(\"height\"),\n loading_attr=row.get(\"loading_attr\"), decoding_attr=row.get(\"decoding_attr\"),\n http_status=row.get(\"http_status\"), content_length=row.get(\"content_length\"),\n )\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}