Fix Python audit findings: configurable http_timeout, scrape id guard, urllib.parse.quote, non-iterable keywords, CI installs wheel + twine check

Author: manchungkali

.github/workflows/ci.yml

@@ -16,4 +16,9 @@ jobs:
       - uses: actions/setup-python@v5
         with:
           python-version: ${{ matrix.python }}
-      - run: python -m unittest discover -s tests -v
+      - run: pip install build twine
+      - run: python -m build
+      - run: twine check dist/*
+      - run: pip install dist/*.whl
+      # Run tests against the installed wheel, not the source tree
+      - run: mv src src_moved && python -m unittest discover -s tests -v

src/gmapsscraper/__init__.py

@@ -13,6 +13,7 @@
 import time
 import urllib.error
 import urllib.request
+from urllib.parse import quote
 from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union
 
 __version__ = "0.1.0"
@@ -41,15 +42,17 @@ def __init__(self, message: str, status: Optional[int] = None, body: Any = None)
         self.body = body
 
 
-def _default_http(method: str, url: str, headers: Dict[str, str], body: Optional[str]) -> Tuple[int, str]:
+def _default_http(
+    method: str, url: str, headers: Dict[str, str], body: Optional[str], timeout: float = 120.0
+) -> Tuple[int, str]:
     request = urllib.request.Request(
         url,
         data=body.encode("utf-8") if body is not None else None,
         headers=headers,
         method=method,
     )
     try:
-        with urllib.request.urlopen(request, timeout=120) as response:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
             return response.getcode(), response.read().decode("utf-8")
     except urllib.error.HTTPError as exc:
         return exc.code, exc.read().decode("utf-8", "replace")
@@ -61,6 +64,8 @@ class GMapsScraper:
     Args:
         api_key: Your API key — get one at https://gmapsscraper.io/dashboard
         base_url: API base URL (default: https://gmapsscraper.io/api/v1).
+        http_timeout: Per-request HTTP timeout in seconds for the default
+            transport (default: 120). Independent of the job-polling ``timeout``.
         http: Optional transport override for testing:
             ``(method, url, headers, body) -> (status_code, response_text)``.
     """
@@ -70,13 +75,16 @@ def __init__(
         api_key: str,
         *,
         base_url: str = DEFAULT_BASE_URL,
+        http_timeout: float = 120.0,
         http: Optional[HttpCallable] = None,
     ):
         if not isinstance(api_key, str) or not api_key.strip():
             raise TypeError("api_key is required — get one at https://gmapsscraper.io/dashboard")
         self._api_key = api_key
         self.base_url = base_url.rstrip("/")
-        self._http = http or _default_http
+        self._http = http or (
+            lambda method, url, headers, body: _default_http(method, url, headers, body, timeout=http_timeout)
+        )
 
     def __repr__(self) -> str:  # never expose the API key in logs/repr
         return f"GMapsScraper(base_url={self.base_url!r})"
@@ -127,7 +135,13 @@ def create_job(self, keywords: Union[str, Iterable[str]], **options: Any) -> Dic
         Returns:
             ``{"id": "job_xxx", "credits_remaining": 8}``
         """
-        kw = [keywords] if isinstance(keywords, str) else list(keywords)
+        try:
+            kw = [keywords] if isinstance(keywords, str) else list(keywords)
+        except TypeError:
+            raise TypeError(
+                'keywords must be a non-empty string or an iterable of non-empty strings, '
+                'e.g. "coffee shop in Austin TX"'
+            ) from None
         if not kw or any(not isinstance(k, str) or not k.strip() for k in kw):
             raise TypeError(
                 'keywords must be a non-empty string or an iterable of non-empty strings, '
@@ -137,7 +151,7 @@ def create_job(self, keywords: Union[str, Iterable[str]], **options: Any) -> Dic
 
     def get_job(self, job_id: str) -> Job:
         """Get job status: ``{"id", "status": "running"|"complete"|"failed", "name"}``."""
-        return self._request(f"/jobs/{urllib.request.quote(job_id, safe='')}")
+        return self._request(f"/jobs/{quote(job_id, safe='')}")
 
     def wait_for_job(
         self,
@@ -170,8 +184,11 @@ def download_csv(self, job_id: str) -> str:
 
         Columns: title, address, phone, email, website, rating, reviews_count,
         category, latitude, longitude, google_maps_url, opening_hours.
+
+        Note: the full CSV is buffered in memory; typical result sets are a few
+        hundred KB at most.
         """
-        return self._request(f"/jobs/{urllib.request.quote(job_id, safe='')}/download", raw=True)
+        return self._request(f"/jobs/{quote(job_id, safe='')}/download", raw=True)
 
     def download_records(self, job_id: str) -> List[BusinessRecord]:
         """Download job results parsed into a list of dicts (one per business)."""
@@ -188,8 +205,11 @@ def scrape(
     ) -> List[BusinessRecord]:
         """One call: create a job, wait for completion, return parsed records."""
         job = self.create_job(keywords, **options)
-        self.wait_for_job(job["id"], poll_interval=poll_interval, timeout=timeout, on_progress=on_progress)
-        return self.download_records(job["id"])
+        job_id = job.get("id") if isinstance(job, dict) else None
+        if not job_id:
+            raise GMapsScraperError("Malformed response from /scrape: missing job id", body=job)
+        self.wait_for_job(job_id, poll_interval=poll_interval, timeout=timeout, on_progress=on_progress)
+        return self.download_records(job_id)
 
     def credits(self) -> Dict[str, Any]:
         """Get remaining credit balance: ``{"credits": 8}``."""

tests/test_client.py

@@ -134,8 +134,17 @@ def http(method, url, headers, body):
                 return 200, csv_text
             return 200, json.dumps({"id": "job_9", "status": "complete"})
 
-        client = GMapsScraper("key_123", http=http)
+        bodies = []
+
+        def recording_http(method, url, headers, body):
+            if body is not None:
+                bodies.append(json.loads(body))
+            return http(method, url, headers, body)
+
+        client = GMapsScraper("key_123", http=recording_http)
         records = client.scrape(["pizza in NYC"], poll_interval=0.001, email=True)
+        # Client-side polling options must never leak into the API request body
+        self.assertEqual(bodies, [{"email": True, "keywords": ["pizza in NYC"]}])
         self.assertEqual(len(records), 2)
         self.assertEqual(records[0]["title"], 'Joe\'s "Best" Pizza, Inc')
         self.assertEqual(records[0]["email"], "joe@example.com")