{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/url_repository.py",
"content": "from __future__ import annotations\n\nfrom dataclasses import replace\n\nfrom src.domain.entities.url import Url\nfrom src.domain.ports.url_repository_port import UrlRepositoryPort\nfrom src.domain.value_objects.normalized_url import NormalizedUrl\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\nfrom src.infrastructure.persistence.mappers.url_mapper import row_to_url, url_to_row\n\n_UPSERT = (\n \"INSERT INTO urls (url_hash, url, scheme, host, path, is_internal,\"\n \" first_seen_at, last_seen_at)\"\n \" VALUES (%(url_hash)s, %(url)s, %(scheme)s, %(host)s, %(path)s, %(is_internal)s,\"\n \" %(first_seen_at)s, %(last_seen_at)s)\"\n \" ON DUPLICATE KEY UPDATE last_seen_at=VALUES(last_seen_at), id=LAST_INSERT_ID(id)\"\n)\n_SELECT_HASH = \"SELECT * FROM urls WHERE url_hash=%s\"\n\n\nclass UrlRepository(UrlRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def upsert(self, url: Url) -> Url:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_UPSERT, url_to_row(url))\n cur.execute(_SELECT_HASH, (url.normalized.url_hash,))\n row = cur.fetchone()\n if row is None:\n raise RuntimeError(\"url upsert failed to return row\")\n return replace(url, id=int(row[\"id\"]), last_seen_at=row[\"last_seen_at\"])\n\n def find_by_hash(self, url_hash: str) -> Url | None:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_HASH, (url_hash,))\n row = cur.fetchone()\n return row_to_url(row) if row else None\n\n def find_by_normalized(self, normalized: NormalizedUrl) -> Url | None:\n return self.find_by_hash(normalized.url_hash)\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/url_repository.py",
"content": "from __future__ import annotations\n\nfrom dataclasses import replace\n\nfrom src.domain.entities.url import Url\nfrom src.domain.ports.url_repository_port import UrlRepositoryPort\nfrom src.domain.value_objects.normalized_url import NormalizedUrl\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\nfrom src.infrastructure.persistence.mappers.url_mapper import row_to_url, url_to_row\n\n_UPSERT = (\n \"INSERT INTO urls (url_hash, url, scheme, host, path, is_internal,\"\n \" first_seen_at, last_seen_at)\"\n \" VALUES (%(url_hash)s, %(url)s, %(scheme)s, %(host)s, %(path)s, %(is_internal)s,\"\n \" %(first_seen_at)s, %(last_seen_at)s)\"\n \" ON DUPLICATE KEY UPDATE last_seen_at=VALUES(last_seen_at), id=LAST_INSERT_ID(id)\"\n)\n_SELECT_HASH = \"SELECT * FROM urls WHERE url_hash=%s\"\n\n\nclass UrlRepository(UrlRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def upsert(self, url: Url) -> Url:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_UPSERT, url_to_row(url))\n cur.execute(_SELECT_HASH, (url.normalized.url_hash,))\n row = cur.fetchone()\n if row is None:\n raise RuntimeError(\"url upsert failed to return row\")\n return replace(url, id=int(row[\"id\"]), last_seen_at=row[\"last_seen_at\"])\n\n def find_by_hash(self, url_hash: str) -> Url | None:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_HASH, (url_hash,))\n row = cur.fetchone()\n return row_to_url(row) if row else None\n\n def find_by_normalized(self, normalized: NormalizedUrl) -> Url | None:\n return self.find_by_hash(normalized.url_hash)\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}