{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/crawl_error_repository.py",
"content": "from __future__ import annotations\n\nfrom dataclasses import replace\nfrom typing import Any\n\nfrom src.domain.entities.crawl_error import CrawlError\nfrom src.domain.ports.crawl_error_repository_port import CrawlErrorRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_INSERT = (\n \"INSERT INTO errors (crawl_id, url, error_type, error_msg, occurred_at)\"\n \" VALUES (%(crawl_id)s, %(url)s, %(error_type)s, %(error_msg)s, %(occurred_at)s)\"\n)\n_SELECT_BY_CRAWL = \"SELECT * FROM errors WHERE crawl_id=%s ORDER BY id\"\n_COUNT_BY_CRAWL = \"SELECT COUNT(*) AS c FROM errors WHERE crawl_id=%s\"\n\n\nclass CrawlErrorRepository(CrawlErrorRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save(self, error: CrawlError) -> CrawlError:\n with self._connections.session() as conn:\n cur = conn.cursor()\n cur.execute(_INSERT, _to_row(error))\n return replace(error, id=int(cur.lastrowid))\n\n def list_by_crawl(self, crawl_id: int) -> list[CrawlError]:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_CRAWL, (crawl_id,))\n return [_to_entity(r) for r in cur.fetchall()]\n\n def count_by_crawl(self, crawl_id: int) -> int:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_COUNT_BY_CRAWL, (crawl_id,))\n row = cur.fetchone()\n return int(row[\"c\"]) if row else 0\n\n\ndef _to_row(e: CrawlError) -> dict[str, Any]:\n return {\n \"crawl_id\": e.crawl_id, \"url\": e.url, \"error_type\": e.error_type,\n \"error_msg\": e.error_msg, \"occurred_at\": e.occurred_at,\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> CrawlError:\n return CrawlError(\n id=row[\"id\"], crawl_id=row[\"crawl_id\"], url=row[\"url\"],\n error_type=row[\"error_type\"], error_msg=row[\"error_msg\"],\n occurred_at=row[\"occurred_at\"],\n )\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/crawl_error_repository.py",
"content": "from __future__ import annotations\n\nfrom dataclasses import replace\nfrom typing import Any\n\nfrom src.domain.entities.crawl_error import CrawlError\nfrom src.domain.ports.crawl_error_repository_port import CrawlErrorRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_INSERT = (\n \"INSERT INTO errors (crawl_id, url, error_type, error_msg, occurred_at)\"\n \" VALUES (%(crawl_id)s, %(url)s, %(error_type)s, %(error_msg)s, %(occurred_at)s)\"\n)\n_SELECT_BY_CRAWL = \"SELECT * FROM errors WHERE crawl_id=%s ORDER BY id\"\n_COUNT_BY_CRAWL = \"SELECT COUNT(*) AS c FROM errors WHERE crawl_id=%s\"\n\n\nclass CrawlErrorRepository(CrawlErrorRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save(self, error: CrawlError) -> CrawlError:\n with self._connections.session() as conn:\n cur = conn.cursor()\n cur.execute(_INSERT, _to_row(error))\n return replace(error, id=int(cur.lastrowid))\n\n def list_by_crawl(self, crawl_id: int) -> list[CrawlError]:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_CRAWL, (crawl_id,))\n return [_to_entity(r) for r in cur.fetchall()]\n\n def count_by_crawl(self, crawl_id: int) -> int:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_COUNT_BY_CRAWL, (crawl_id,))\n row = cur.fetchone()\n return int(row[\"c\"]) if row else 0\n\n\ndef _to_row(e: CrawlError) -> dict[str, Any]:\n return {\n \"crawl_id\": e.crawl_id, \"url\": e.url, \"error_type\": e.error_type,\n \"error_msg\": e.error_msg, \"occurred_at\": e.occurred_at,\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> CrawlError:\n return CrawlError(\n id=row[\"id\"], crawl_id=row[\"crawl_id\"], url=row[\"url\"],\n error_type=row[\"error_type\"], error_msg=row[\"error_msg\"],\n occurred_at=row[\"occurred_at\"],\n )\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}