{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/resource_repository.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterable\nfrom typing import Any\n\nfrom src.domain.entities.resource import Resource\nfrom src.domain.ports.resource_repository_port import ResourceRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_COLUMNS = (\n \"page_id, resource_type, url, http_status, content_length, duration_ms, from_cache\"\n)\n_PLACEHOLDERS = \", \".join(f\"%({n.strip()})s\" for n in _COLUMNS.split(\",\"))\n_INSERT = f\"INSERT INTO resources ({_COLUMNS}) VALUES ({_PLACEHOLDERS})\"\n_SELECT_BY_PAGE = \"SELECT * FROM resources WHERE page_id=%s ORDER BY id\"\n\n\nclass ResourceRepository(ResourceRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, resources: Iterable[Resource]) -> None:\n rows = [_to_row(r) for r in resources]\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[Resource]:\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(r: Resource) -> dict[str, Any]:\n return {\n \"page_id\": r.page_id, \"resource_type\": r.resource_type, \"url\": r.url,\n \"http_status\": r.http_status, \"content_length\": r.content_length,\n \"duration_ms\": r.duration_ms,\n \"from_cache\": None if r.from_cache is None else int(r.from_cache),\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> Resource:\n cached = row.get(\"from_cache\")\n return Resource(\n id=row[\"id\"], page_id=row[\"page_id\"], resource_type=row[\"resource_type\"],\n url=row[\"url\"], http_status=row.get(\"http_status\"),\n content_length=row.get(\"content_length\"),\n duration_ms=row.get(\"duration_ms\"),\n from_cache=bool(cached) if cached is not None else None,\n )\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/resource_repository.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Iterable\nfrom typing import Any\n\nfrom src.domain.entities.resource import Resource\nfrom src.domain.ports.resource_repository_port import ResourceRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_COLUMNS = (\n \"page_id, resource_type, url, http_status, content_length, duration_ms, from_cache\"\n)\n_PLACEHOLDERS = \", \".join(f\"%({n.strip()})s\" for n in _COLUMNS.split(\",\"))\n_INSERT = f\"INSERT INTO resources ({_COLUMNS}) VALUES ({_PLACEHOLDERS})\"\n_SELECT_BY_PAGE = \"SELECT * FROM resources WHERE page_id=%s ORDER BY id\"\n\n\nclass ResourceRepository(ResourceRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, resources: Iterable[Resource]) -> None:\n rows = [_to_row(r) for r in resources]\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[Resource]:\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(r: Resource) -> dict[str, Any]:\n return {\n \"page_id\": r.page_id, \"resource_type\": r.resource_type, \"url\": r.url,\n \"http_status\": r.http_status, \"content_length\": r.content_length,\n \"duration_ms\": r.duration_ms,\n \"from_cache\": None if r.from_cache is None else int(r.from_cache),\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> Resource:\n cached = row.get(\"from_cache\")\n return Resource(\n id=row[\"id\"], page_id=row[\"page_id\"], resource_type=row[\"resource_type\"],\n url=row[\"url\"], http_status=row.get(\"http_status\"),\n content_length=row.get(\"content_length\"),\n duration_ms=row.get(\"duration_ms\"),\n from_cache=bool(cached) if cached is not None else None,\n )\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}