{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/fetchers\/httpx_fetcher.py",
"content": "from __future__ import annotations\n\nimport time\nfrom typing import Self\n\nimport httpx\n\nfrom src.domain.ports.http_fetcher_port import FetchResult, HttpFetcherPort\nfrom src.domain.value_objects.normalized_url import NormalizedUrl\nfrom src.infrastructure.config.constants import HTTP_TIMEOUT_SEC, USER_AGENT\n\n_MS_PER_SECOND = 1000\n\n\nclass HttpxFetcher(HttpFetcherPort):\n def __init__(self, client: httpx.AsyncClient) -> None:\n self._client = client\n\n async def fetch(self, url: NormalizedUrl) -> FetchResult:\n start = time.perf_counter()\n try:\n response = await self._client.get(url.url)\n except httpx.HTTPError as exc:\n elapsed = int((time.perf_counter() - start) * _MS_PER_SECOND)\n return _error_result(url, str(exc), elapsed)\n elapsed_ms = int((time.perf_counter() - start) * _MS_PER_SECOND)\n return _success_result(response, elapsed_ms)\n\n @classmethod\n def create(cls) -> Self:\n return cls(httpx.AsyncClient(\n timeout=HTTP_TIMEOUT_SEC,\n follow_redirects=True,\n headers={\"User-Agent\": USER_AGENT},\n ))\n\n async def close(self) -> None:\n await self._client.aclose()\n\n\ndef _success_result(response: httpx.Response, elapsed_ms: int) -> FetchResult:\n chain = tuple(str(r.url) for r in response.history)\n return FetchResult(\n final_url=str(response.url),\n status_code=response.status_code,\n redirect_chain=chain,\n headers={k.lower(): v for k, v in response.headers.items()},\n body=response.content,\n response_time_ms=elapsed_ms,\n ttfb_ms=None,\n error=None,\n )\n\n\ndef _error_result(url: NormalizedUrl, message: str, elapsed_ms: int) -> FetchResult:\n return FetchResult(\n final_url=url.url,\n status_code=0,\n redirect_chain=(),\n headers={},\n body=b\"\",\n response_time_ms=elapsed_ms,\n ttfb_ms=None,\n error=message,\n )\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/infrastructure\/fetchers\/httpx_fetcher.py",
"content": "from __future__ import annotations\n\nimport time\nfrom typing import Self\n\nimport httpx\n\nfrom src.domain.ports.http_fetcher_port import FetchResult, HttpFetcherPort\nfrom src.domain.value_objects.normalized_url import NormalizedUrl\nfrom src.infrastructure.config.constants import HTTP_TIMEOUT_SEC, USER_AGENT\n\n_MS_PER_SECOND = 1000\n\n\nclass HttpxFetcher(HttpFetcherPort):\n def __init__(self, client: httpx.AsyncClient) -> None:\n self._client = client\n\n async def fetch(self, url: NormalizedUrl) -> FetchResult:\n start = time.perf_counter()\n try:\n response = await self._client.get(url.url)\n except httpx.HTTPError as exc:\n elapsed = int((time.perf_counter() - start) * _MS_PER_SECOND)\n return _error_result(url, str(exc), elapsed)\n elapsed_ms = int((time.perf_counter() - start) * _MS_PER_SECOND)\n return _success_result(response, elapsed_ms)\n\n @classmethod\n def create(cls) -> Self:\n return cls(httpx.AsyncClient(\n timeout=HTTP_TIMEOUT_SEC,\n follow_redirects=True,\n headers={\"User-Agent\": USER_AGENT},\n ))\n\n async def close(self) -> None:\n await self._client.aclose()\n\n\ndef _success_result(response: httpx.Response, elapsed_ms: int) -> FetchResult:\n chain = tuple(str(r.url) for r in response.history)\n return FetchResult(\n final_url=str(response.url),\n status_code=response.status_code,\n redirect_chain=chain,\n headers={k.lower(): v for k, v in response.headers.items()},\n body=response.content,\n response_time_ms=elapsed_ms,\n ttfb_ms=None,\n error=None,\n )\n\n\ndef _error_result(url: NormalizedUrl, message: str, elapsed_ms: int) -> FetchResult:\n return FetchResult(\n final_url=url.url,\n status_code=0,\n redirect_chain=(),\n headers={},\n body=b\"\",\n response_time_ms=elapsed_ms,\n ttfb_ms=None,\n error=message,\n )\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}