{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/external_check_repository.py",
"content": "from __future__ import annotations\n\nfrom dataclasses import replace\nfrom typing import Any\n\nfrom src.domain.entities.external_check import ExternalCheck\nfrom src.domain.ports.external_check_repository_port import ExternalCheckRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_UPSERT = (\n \"INSERT INTO external_checks\"\n \" (url_hash, url, http_status, final_url, response_time_ms, checked_at, error)\"\n \" VALUES (%(url_hash)s, %(url)s, %(http_status)s, %(final_url)s,\"\n \" %(response_time_ms)s, %(checked_at)s, %(error)s)\"\n \" ON DUPLICATE KEY UPDATE http_status=VALUES(http_status),\"\n \" final_url=VALUES(final_url), response_time_ms=VALUES(response_time_ms),\"\n \" checked_at=VALUES(checked_at), error=VALUES(error), id=LAST_INSERT_ID(id)\"\n)\n_SELECT_BY_HASH = \"SELECT * FROM external_checks WHERE url_hash=%s\"\n\n\nclass ExternalCheckRepository(ExternalCheckRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def upsert(self, check: ExternalCheck) -> ExternalCheck:\n with self._connections.session() as conn:\n cur = conn.cursor()\n cur.execute(_UPSERT, _to_row(check))\n return replace(check, id=int(cur.lastrowid))\n\n def get_by_hash(self, url_hash: str) -> ExternalCheck | None:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_HASH, (url_hash,))\n row = cur.fetchone()\n return _to_entity(row) if row else None\n\n\ndef _to_row(c: ExternalCheck) -> dict[str, Any]:\n return {\n \"url_hash\": c.url_hash, \"url\": c.url, \"http_status\": c.http_status,\n \"final_url\": c.final_url, \"response_time_ms\": c.response_time_ms,\n \"checked_at\": c.checked_at, \"error\": c.error,\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> ExternalCheck:\n return ExternalCheck(\n id=row[\"id\"], url_hash=row[\"url_hash\"], url=row[\"url\"],\n http_status=row.get(\"http_status\"), final_url=row.get(\"final_url\"),\n response_time_ms=row.get(\"response_time_ms\"), checked_at=row[\"checked_at\"],\n error=row.get(\"error\"),\n )\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/external_check_repository.py",
"content": "from __future__ import annotations\n\nfrom dataclasses import replace\nfrom typing import Any\n\nfrom src.domain.entities.external_check import ExternalCheck\nfrom src.domain.ports.external_check_repository_port import ExternalCheckRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_UPSERT = (\n \"INSERT INTO external_checks\"\n \" (url_hash, url, http_status, final_url, response_time_ms, checked_at, error)\"\n \" VALUES (%(url_hash)s, %(url)s, %(http_status)s, %(final_url)s,\"\n \" %(response_time_ms)s, %(checked_at)s, %(error)s)\"\n \" ON DUPLICATE KEY UPDATE http_status=VALUES(http_status),\"\n \" final_url=VALUES(final_url), response_time_ms=VALUES(response_time_ms),\"\n \" checked_at=VALUES(checked_at), error=VALUES(error), id=LAST_INSERT_ID(id)\"\n)\n_SELECT_BY_HASH = \"SELECT * FROM external_checks WHERE url_hash=%s\"\n\n\nclass ExternalCheckRepository(ExternalCheckRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def upsert(self, check: ExternalCheck) -> ExternalCheck:\n with self._connections.session() as conn:\n cur = conn.cursor()\n cur.execute(_UPSERT, _to_row(check))\n return replace(check, id=int(cur.lastrowid))\n\n def get_by_hash(self, url_hash: str) -> ExternalCheck | None:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_HASH, (url_hash,))\n row = cur.fetchone()\n return _to_entity(row) if row else None\n\n\ndef _to_row(c: ExternalCheck) -> dict[str, Any]:\n return {\n \"url_hash\": c.url_hash, \"url\": c.url, \"http_status\": c.http_status,\n \"final_url\": c.final_url, \"response_time_ms\": c.response_time_ms,\n \"checked_at\": c.checked_at, \"error\": c.error,\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> ExternalCheck:\n return ExternalCheck(\n id=row[\"id\"], url_hash=row[\"url_hash\"], url=row[\"url\"],\n http_status=row.get(\"http_status\"), final_url=row.get(\"final_url\"),\n response_time_ms=row.get(\"response_time_ms\"), checked_at=row[\"checked_at\"],\n error=row.get(\"error\"),\n )\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}