{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/page_repository.py",
"content": "from __future__ import annotations\n\nfrom dataclasses import replace\n\nfrom src.domain.entities.page import Page\nfrom src.domain.ports.page_repository_port import PageRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\nfrom src.infrastructure.persistence.mappers.page_mapper import page_to_row, row_to_page\n\n_COLUMNS = (\n \"crawl_id, url_id, fetched_at, render_mode, http_status, final_url, redirect_chain,\"\n \" response_time_ms, ttfb_ms, content_type, content_length, content_encoding,\"\n \" title, title_length, meta_description, meta_description_length, meta_keywords,\"\n \" meta_robots, canonical, lang, charset, viewport,\"\n \" og_title, og_description, og_image, og_type, og_url, og_site_name, og_locale,\"\n \" twitter_card, twitter_title, twitter_description, twitter_image,\"\n \" h1_count, h1_texts, h2_count, h2_texts, h3_count, h4_count, h5_count, h6_count,\"\n \" word_count, text_hash, dom_node_count, render_time_ms, lcp_ms, cls, tbt_ms,\"\n \" screenshot_path, html_raw_path, quality_flags\"\n)\n_PLACEHOLDERS = \", \".join(f\"%({name.strip()})s\" for name in _COLUMNS.split(\",\"))\n_INSERT = f\"INSERT INTO pages ({_COLUMNS}) VALUES ({_PLACEHOLDERS})\"\n_SELECT_BY_ID = \"SELECT * FROM pages WHERE id=%s\"\n_SELECT_BY_CRAWL = (\n \"SELECT * FROM pages WHERE crawl_id=%s ORDER BY id LIMIT %s OFFSET %s\"\n)\n\n\nclass PageRepository(PageRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save(self, page: Page) -> Page:\n with self._connections.session() as conn:\n cur = conn.cursor()\n cur.execute(_INSERT, page_to_row(page))\n return replace(page, id=int(cur.lastrowid))\n\n def get(self, page_id: int) -> Page | None:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_ID, (page_id,))\n row = cur.fetchone()\n return row_to_page(row) if row else None\n\n def list_by_crawl(self, crawl_id: int, limit: int, offset: int) -> list[Page]:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_CRAWL, (crawl_id, limit, offset))\n return [row_to_page(r) for r in cur.fetchall()]\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/persistence\/page_repository.py",
"content": "from __future__ import annotations\n\nfrom dataclasses import replace\n\nfrom src.domain.entities.page import Page\nfrom src.domain.ports.page_repository_port import PageRepositoryPort\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\nfrom src.infrastructure.persistence.mappers.page_mapper import page_to_row, row_to_page\n\n_COLUMNS = (\n \"crawl_id, url_id, fetched_at, render_mode, http_status, final_url, redirect_chain,\"\n \" response_time_ms, ttfb_ms, content_type, content_length, content_encoding,\"\n \" title, title_length, meta_description, meta_description_length, meta_keywords,\"\n \" meta_robots, canonical, lang, charset, viewport,\"\n \" og_title, og_description, og_image, og_type, og_url, og_site_name, og_locale,\"\n \" twitter_card, twitter_title, twitter_description, twitter_image,\"\n \" h1_count, h1_texts, h2_count, h2_texts, h3_count, h4_count, h5_count, h6_count,\"\n \" word_count, text_hash, dom_node_count, render_time_ms, lcp_ms, cls, tbt_ms,\"\n \" screenshot_path, html_raw_path, quality_flags\"\n)\n_PLACEHOLDERS = \", \".join(f\"%({name.strip()})s\" for name in _COLUMNS.split(\",\"))\n_INSERT = f\"INSERT INTO pages ({_COLUMNS}) VALUES ({_PLACEHOLDERS})\"\n_SELECT_BY_ID = \"SELECT * FROM pages WHERE id=%s\"\n_SELECT_BY_CRAWL = (\n \"SELECT * FROM pages WHERE crawl_id=%s ORDER BY id LIMIT %s OFFSET %s\"\n)\n\n\nclass PageRepository(PageRepositoryPort):\n def __init__(self, connections: MariaDbConnectionFactory) -> None:\n self._connections = connections\n\n def save(self, page: Page) -> Page:\n with self._connections.session() as conn:\n cur = conn.cursor()\n cur.execute(_INSERT, page_to_row(page))\n return replace(page, id=int(cur.lastrowid))\n\n def get(self, page_id: int) -> Page | None:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_ID, (page_id,))\n row = cur.fetchone()\n return row_to_page(row) if row else None\n\n def list_by_crawl(self, crawl_id: int, limit: int, offset: int) -> list[Page]:\n with self._connections.session() as conn:\n cur = conn.cursor(dictionary=True)\n cur.execute(_SELECT_BY_CRAWL, (crawl_id, limit, offset))\n return [row_to_page(r) for r in cur.fetchall()]\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}