Centralize paginated web job response initialization

Co-authored-by: Shri Sukhani <shrisukhani@users.noreply.github.com>

Author: cursoragent

hyperbrowser/client/managers/async_manager/web/batch_fetch.py

@@ -10,6 +10,7 @@
 )
 from ...web_payload_utils import build_batch_fetch_start_payload
 from ...web_payload_utils import build_batch_fetch_get_params
+from ...web_pagination_utils import initialize_paginated_job_response
 from ....polling import (
     build_fetch_operation_name,
     build_operation_name,
@@ -96,14 +97,10 @@ async def start_and_wait(
                 retry_delay_seconds=0.5,
             )
 
-        job_response = BatchFetchJobResponse(
-            jobId=job_id,
+        job_response = initialize_paginated_job_response(
+            model=BatchFetchJobResponse,
+            job_id=job_id,
             status=job_status,
-            data=[],
-            currentPageBatch=0,
-            totalPageBatches=0,
-            totalPages=0,
-            batchSize=100,
         )
 
         def merge_page_response(page_response: BatchFetchJobResponse) -> None:

hyperbrowser/client/managers/async_manager/web/crawl.py

@@ -10,6 +10,7 @@
 )
 from ...web_payload_utils import build_web_crawl_start_payload
 from ...web_payload_utils import build_web_crawl_get_params
+from ...web_pagination_utils import initialize_paginated_job_response
 from ....polling import (
     build_fetch_operation_name,
     build_operation_name,
@@ -94,14 +95,10 @@ async def start_and_wait(
                 retry_delay_seconds=0.5,
             )
 
-        job_response = WebCrawlJobResponse(
-            jobId=job_id,
+        job_response = initialize_paginated_job_response(
+            model=WebCrawlJobResponse,
+            job_id=job_id,
             status=job_status,
-            data=[],
-            currentPageBatch=0,
-            totalPageBatches=0,
-            totalPages=0,
-            batchSize=100,
         )
 
         def merge_page_response(page_response: WebCrawlJobResponse) -> None:

hyperbrowser/client/managers/sync_manager/web/batch_fetch.py

@@ -10,6 +10,7 @@
 )
 from ...web_payload_utils import build_batch_fetch_start_payload
 from ...web_payload_utils import build_batch_fetch_get_params
+from ...web_pagination_utils import initialize_paginated_job_response
 from ....polling import (
     build_fetch_operation_name,
     build_operation_name,
@@ -94,14 +95,10 @@ def start_and_wait(
                 retry_delay_seconds=0.5,
             )
 
-        job_response = BatchFetchJobResponse(
-            jobId=job_id,
+        job_response = initialize_paginated_job_response(
+            model=BatchFetchJobResponse,
+            job_id=job_id,
             status=job_status,
-            data=[],
-            currentPageBatch=0,
-            totalPageBatches=0,
-            totalPages=0,
-            batchSize=100,
         )
 
         def merge_page_response(page_response: BatchFetchJobResponse) -> None:

hyperbrowser/client/managers/sync_manager/web/crawl.py

@@ -10,6 +10,7 @@
 )
 from ...web_payload_utils import build_web_crawl_start_payload
 from ...web_payload_utils import build_web_crawl_get_params
+from ...web_pagination_utils import initialize_paginated_job_response
 from ....polling import (
     build_fetch_operation_name,
     build_operation_name,
@@ -94,14 +95,10 @@ def start_and_wait(
                 retry_delay_seconds=0.5,
             )
 
-        job_response = WebCrawlJobResponse(
-            jobId=job_id,
+        job_response = initialize_paginated_job_response(
+            model=WebCrawlJobResponse,
+            job_id=job_id,
             status=job_status,
-            data=[],
-            currentPageBatch=0,
-            totalPageBatches=0,
-            totalPages=0,
-            batchSize=100,
         )
 
         def merge_page_response(page_response: WebCrawlJobResponse) -> None:

hyperbrowser/client/managers/web_pagination_utils.py

@@ -0,0 +1,21 @@
+from typing import Type, TypeVar
+
+T = TypeVar("T")
+
+
+def initialize_paginated_job_response(
+    *,
+    model: Type[T],
+    job_id: str,
+    status: str,
+    batch_size: int = 100,
+) -> T:
+    return model(
+        jobId=job_id,
+        status=status,
+        data=[],
+        currentPageBatch=0,
+        totalPageBatches=0,
+        totalPages=0,
+        batchSize=batch_size,
+    )

tests/test_web_pagination_utils.py

@@ -0,0 +1,37 @@
+from hyperbrowser.client.managers.web_pagination_utils import (
+    initialize_paginated_job_response,
+)
+from hyperbrowser.models import BatchFetchJobResponse, WebCrawlJobResponse
+
+
+def test_initialize_paginated_job_response_for_batch_fetch():
+    response = initialize_paginated_job_response(
+        model=BatchFetchJobResponse,
+        job_id="job-1",
+        status="completed",
+    )
+
+    assert response.job_id == "job-1"
+    assert response.status == "completed"
+    assert response.data == []
+    assert response.current_page_batch == 0
+    assert response.total_page_batches == 0
+    assert response.total_pages == 0
+    assert response.batch_size == 100
+
+
+def test_initialize_paginated_job_response_for_web_crawl_with_custom_batch_size():
+    response = initialize_paginated_job_response(
+        model=WebCrawlJobResponse,
+        job_id="job-2",
+        status="running",
+        batch_size=25,
+    )
+
+    assert response.job_id == "job-2"
+    assert response.status == "running"
+    assert response.data == []
+    assert response.current_page_batch == 0
+    assert response.total_page_batches == 0
+    assert response.total_pages == 0
+    assert response.batch_size == 25

tests/test_web_payload_helper_usage.py

@@ -35,14 +35,18 @@ def test_batch_fetch_managers_use_shared_start_payload_helper():
         module_text = Path(module_path).read_text(encoding="utf-8")
         assert "build_batch_fetch_start_payload(" in module_text
         assert "build_batch_fetch_get_params(" in module_text
+        assert "initialize_paginated_job_response(" in module_text
         assert "inject_web_output_schemas(" not in module_text
         assert "serialize_model_dump_to_dict(" not in module_text
+        assert "BatchFetchJobResponse(" not in module_text
 
 
 def test_web_crawl_managers_use_shared_start_payload_helper():
     for module_path in WEB_CRAWL_MANAGER_MODULES:
         module_text = Path(module_path).read_text(encoding="utf-8")
         assert "build_web_crawl_start_payload(" in module_text
         assert "build_web_crawl_get_params(" in module_text
+        assert "initialize_paginated_job_response(" in module_text
         assert "inject_web_output_schemas(" not in module_text
         assert "serialize_model_dump_to_dict(" not in module_text
+        assert "WebCrawlJobResponse(" not in module_text