{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/structured_data_repository.py",
"content": "from __future__ import annotations\n\nimport json\nfrom collections.abc import Iterable\nfrom typing import Any\n\nfrom src.domain.entities.structured_data_item import StructuredDataItem\nfrom src.domain.ports.structured_data_repository_port import StructuredDataRepositoryPort\nfrom src.domain.value_objects.enums import StructuredDataFormat\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_INSERT = (\n \"INSERT INTO structured_data (page_id, format, schema_type, payload)\"\n \" VALUES (%(page_id)s, %(format)s, %(schema_type)s, %(payload)s)\"\n)\n_SELECT_BY_PAGE = \"SELECT * FROM structured_data WHERE page_id=%s ORDER BY id\"\n\n\nclass StructuredDataRepository(StructuredDataRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, items: Iterable[StructuredDataItem]) -> None:\n rows = [_to_row(i) for i in items]\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[StructuredDataItem]:\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(item: StructuredDataItem) -> dict[str, Any]:\n return {\n \"page_id\": item.page_id,\n \"format\": item.format.value,\n \"schema_type\": item.schema_type,\n \"payload\": json.dumps(dict(item.payload)),\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> StructuredDataItem:\n return StructuredDataItem(\n id=row[\"id\"], page_id=row[\"page_id\"],\n format=StructuredDataFormat(row[\"format\"]),\n schema_type=row.get(\"schema_type\"),\n payload=json.loads(row[\"payload\"]) if row[\"payload\"] else {},\n )\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/structured_data_repository.py",
"content": "from __future__ import annotations\n\nimport json\nfrom collections.abc import Iterable\nfrom typing import Any\n\nfrom src.domain.entities.structured_data_item import StructuredDataItem\nfrom src.domain.ports.structured_data_repository_port import StructuredDataRepositoryPort\nfrom src.domain.value_objects.enums import StructuredDataFormat\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\n\n_INSERT = (\n \"INSERT INTO structured_data (page_id, format, schema_type, payload)\"\n \" VALUES (%(page_id)s, %(format)s, %(schema_type)s, %(payload)s)\"\n)\n_SELECT_BY_PAGE = \"SELECT * FROM structured_data WHERE page_id=%s ORDER BY id\"\n\n\nclass StructuredDataRepository(StructuredDataRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save_many(self, items: Iterable[StructuredDataItem]) -> None:\n rows = [_to_row(i) for i in items]\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[StructuredDataItem]:\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(item: StructuredDataItem) -> dict[str, Any]:\n return {\n \"page_id\": item.page_id,\n \"format\": item.format.value,\n \"schema_type\": item.schema_type,\n \"payload\": json.dumps(dict(item.payload)),\n }\n\n\ndef _to_entity(row: dict[str, Any]) -> StructuredDataItem:\n return StructuredDataItem(\n id=row[\"id\"], page_id=row[\"page_id\"],\n format=StructuredDataFormat(row[\"format\"]),\n schema_type=row.get(\"schema_type\"),\n payload=json.loads(row[\"payload\"]) if row[\"payload\"] else {},\n )\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}