chore(types): add pyright gate and tighten engine seams

Wire pyright into make check, tighten Settings.sentry and DriverProtocol
seams, and relax test-only diagnostics via executionEnvironments.

Author: gildesmarais

Makefile

@@ -1,4 +1,4 @@
-.PHONY: test build serve health scrape-example smoke lint lintfix check ready openapi openapi-verify spectral
+.PHONY: test build serve health scrape-example smoke lint lintfix check ready openapi openapi-verify spectral typecheck
 
 .DEFAULT_GOAL := check
 
@@ -35,13 +35,16 @@ openapi-verify:
 	$(PYTHON) scripts/export_openapi.py --out $$tmp && \
 	diff -u $(OPENAPI_FILE) $$tmp
 
-check: lint test openapi-verify
+check: lint test typecheck openapi-verify
 
 ready: check
 
 test:
 	$(PYTHON) -m unittest discover -s tests
 
+typecheck:
+	$(PYTHON) -m pyright app tests
+
 
 build:
 	docker build -t $(IMAGE) .

app/api/openapi.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from typing import Any, cast
 
 from app.api.openapi_examples import (
     HEALTH_EXAMPLE,
@@ -17,9 +18,9 @@
 class OpenApiMetadata:
     api_description: str
     json_error_example: dict[str, dict[str, dict]]
-    scrape_error_responses: dict[int, dict]
-    health_responses: dict[int, dict]
-    scrape_success_response: dict[int, dict]
+    scrape_error_responses: dict[int | str, dict[str, Any]]
+    health_responses: dict[int | str, dict[str, Any]]
+    scrape_success_response: dict[int | str, dict[str, Any]]
     openapi_tags: list[dict[str, str]]
     servers: list[dict[str, str]]
     contact: dict[str, str]
@@ -115,7 +116,9 @@ def build_openapi_metadata(settings: Settings) -> OpenApiMetadata:
     return OpenApiMetadata(
         api_description=api_description,
         json_error_example=json_error_example,
-        scrape_error_responses=scrape_error_responses,
+        scrape_error_responses=cast(
+            dict[int | str, dict[str, Any]], scrape_error_responses
+        ),
         health_responses={
             200: {
                 "description": "Service is up.",
@@ -158,9 +161,9 @@ def build_openapi_metadata(settings: Settings) -> OpenApiMetadata:
 # Populated by configure_openapi() during create_app(); route modules import these names.
 API_DESCRIPTION = ""
 JSON_ERROR_EXAMPLE: dict[str, dict[str, dict]] = {}
-SCRAPE_ERROR_RESPONSES: dict[int, dict] = {}
-HEALTH_RESPONSES: dict[int, dict] = {}
-SCRAPE_SUCCESS_RESPONSE: dict[int, dict] = {}
+SCRAPE_ERROR_RESPONSES: dict[int | str, dict[str, Any]] = {}
+HEALTH_RESPONSES: dict[int | str, dict[str, Any]] = {}
+SCRAPE_SUCCESS_RESPONSE: dict[int | str, dict[str, Any]] = {}
 OPENAPI_TAGS: list[dict[str, str]] = []
 SERVERS: list[dict[str, str]] = []
 CONTACT: dict[str, str] = {}

app/config.py

@@ -74,11 +74,10 @@ class Settings(BaseSettings):
     )
     runtime_root: Path = Field(default=Path("/tmp/scrape"))
     environment: str = Field(default="production", validation_alias="ENVIRONMENT")
-    sentry: SentrySettings | None = None
+    sentry: SentrySettings = Field(default_factory=SentrySettings)
 
     @model_validator(mode="after")
     def validate_timeout_relationship(self) -> Settings:
-        object.__setattr__(self, "sentry", SentrySettings())
         if self.scrape_work_timeout_seconds > self.scrape_timeout_seconds:
             raise ValueError(
                 "SCRAPE_WORK_TIMEOUT_SECONDS cannot exceed SCRAPE_TIMEOUT_SECONDS: "

app/engine/browser_tier.py

@@ -5,7 +5,7 @@
 import errno
 import logging
 import time
-from typing import Any
+from typing import Any, cast
 from urllib.parse import urlparse
 
 from app.config import Settings
@@ -113,21 +113,25 @@ def run_browser_tier(
         else None
     )
 
-    session.driver = Driver(
-        headless=payload.headless,
-        enable_xvfb_virtual_display=not payload.headless,
-        proxy=payload.proxy,
-        profile=str(session.profile_dir),
-        tiny_profile=True,
-        block_images=payload.block_images,
-        block_images_and_css=payload.block_images_and_css,
-        wait_for_complete_page_load=payload.wait_for_complete_page_load,
-        user_agent=payload.effective_user_agent,
-        window_size=driver_window_size,
-        lang=payload.lang,
-        remove_default_browser_check_argument=True,
+    driver = cast(
+        DriverProtocol,
+        Driver(
+            headless=payload.headless,
+            enable_xvfb_virtual_display=not payload.headless,
+            proxy=payload.proxy,
+            profile=str(session.profile_dir),
+            tiny_profile=True,
+            block_images=payload.block_images,
+            block_images_and_css=payload.block_images_and_css,
+            wait_for_complete_page_load=payload.wait_for_complete_page_load,
+            user_agent=payload.effective_user_agent,
+            window_size=driver_window_size,
+            lang=payload.lang,
+            remove_default_browser_check_argument=True,
+        ),
     )
-    configure_driver(session.driver, payload, target_url, collector=collector)
+    session.driver = driver
+    configure_driver(driver, payload, target_url, collector=collector)
     browser_ready_monotonic = time.monotonic()
     progress.mark(
         TimeoutPhase.WORK,
@@ -146,33 +150,33 @@ def run_browser_tier(
             step_budget = browser_step_budget_seconds(
                 settings, started_monotonic, browser_ready_monotonic
             )
-            navigate(session.driver, target_url, strategy, step_budget)
+            navigate(driver, target_url, strategy, step_budget)
             wait_for_readiness(
-                session.driver,
+                driver,
                 selector=payload.wait_for_selector,
                 timeout_seconds=min(payload.wait_timeout_seconds, step_budget),
             )
 
             if payload.scroll:
-                apply_scrolling(session.driver)
+                apply_scrolling(driver)
 
             html, meta, assessment, xhr_responses = collect_page_state(
-                session.driver,
+                driver,
                 target_url,
                 collector,
                 include_xhr=False,
             )
 
             if assessment.challenge_detected or assessment.blocked_detected:
-                call_if_available(session.driver, "bypass_cloudflare")
+                call_if_available(driver, "bypass_cloudflare")
                 html, meta, assessment, xhr_responses = collect_page_state(
-                    session.driver,
+                    driver,
                     target_url,
                     collector,
                     include_xhr=True,
                 )
             else:
-                xhr_responses = harvest_xhr(collector, session.driver)
+                xhr_responses = harvest_xhr(collector, driver)
 
             if assessment.challenge_detected or assessment.blocked_detected:
                 logger.warning(

app/engine/driver_capabilities.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Protocol, runtime_checkable
+from typing import Any, Protocol, cast, runtime_checkable
 
 logger = logging.getLogger("botasaurus_scrape_api")
 
@@ -62,7 +62,7 @@ def call_if_available[T](
     if not callable(method):
         return default
     try:
-        return method(*args, **kwargs)
+        return cast(T, method(*args, **kwargs))
     except Exception as exc:
         logger.debug("driver_capability_failed method=%s error=%s", name, exc)
         return default

app/engine/envelope.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, cast
 
 from app.infra.detector import ChallengeAssessment
 from app.schemas.enums import ErrorCategory, ExecutionTier, NavigationMode, TimeoutPhase
@@ -96,7 +96,7 @@ def build_success(
         headers=headers,
         html=html,
         metadata_error=metadata_error,
-        xhr_responses=xhr_responses or [],
+        xhr_responses=cast(list[XhrResponse], xhr_responses or []),
         diagnostics=build_diagnostics(
             request_id=request_id,
             attempts=attempts,

app/infra/request_id.py

@@ -22,7 +22,7 @@ def resolve_request_id(
     @return tuple of resolved id and used_fallback flag
     """
     candidate = inbound.strip() if inbound is not None else None
-    if _is_valid(candidate):
+    if candidate is not None and _is_valid(candidate):
         return candidate, False
 
     reason = "absent" if not candidate else "invalid"

app/security/url_guard.py

@@ -77,7 +77,7 @@ def validate(cls, raw_url: str) -> ValidationResult:
         for info in addr_infos:
             sockaddr = info[4]
             if sockaddr:
-                resolved_ips.add(sockaddr[0])
+                resolved_ips.add(str(sockaddr[0]))
 
         for ip_text in resolved_ips:
             try:

pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
 dev = [
     "ruff==0.16.4",
     "PyYAML==6.0.3",
+    "pyright==1.1.406",
 ]
 
 [tool.ruff]
@@ -27,3 +28,21 @@ ignore = [
     "E501",
     "SIM105",
 ]
+
+[tool.pyright]
+pythonVersion = "3.14"
+venvPath = "."
+venv = ".venv"
+typeCheckingMode = "standard"
+include = ["app", "tests"]
+exclude = [".venv"]
+reportMissingTypeStubs = false
+reportMissingImports = false
+
+[[tool.pyright.executionEnvironments]]
+root = "tests"
+reportArgumentType = "none"
+reportAttributeAccessIssue = "none"
+reportOptionalSubscript = "none"
+reportOptionalMemberAccess = "none"
+reportFunctionMemberAccess = "none"

tests/support/fakes.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from pathlib import Path
-from typing import ClassVar
+from typing import Any, ClassVar
 
 
 class FakeMetadataResponse:
@@ -51,7 +51,7 @@ def close(self):
 
 
 class CaptureDriver(FakeDriver):
-    last_init_kwargs = None
+    last_init_kwargs: ClassVar[dict[str, Any] | None] = None
 
     def __init__(self, *args, **kwargs):
         type(self).last_init_kwargs = dict(kwargs)
@@ -67,7 +67,7 @@ def __init__(self, *, text, status_code, headers, url):
 
 
 class FakeRequest:
-    response = None
+    response: FakeHttpResponse | None = None
 
     def get(self, *_args, **_kwargs):
         return type(self).response