Skip to content

Transport

HTTP transport layer for vgmdb-client.

Authenticated, Cloudflare-aware fetching of vgmdb pages with sync and async clients over a shared sans-I/O core.

AsyncTransport

Authenticated, Cloudflare-aware asynchronous fetcher for vgmdb pages.

An instance is scoped to a single event loop (its throttle lock and httpx client bind to the loop they are used on); do not share one instance across concurrently-running loops.

Source code in src/vgmdb_client/transport/async_client.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class AsyncTransport:
    """Authenticated, Cloudflare-aware asynchronous fetcher for vgmdb pages.

    An instance is scoped to a single event loop (its throttle lock and httpx client bind to the
    loop they are used on); do not share one instance across concurrently-running loops.
    """

    def __init__(self, config: TransportConfig) -> None:
        self._config = config
        self._last: float | None = None
        self._throttle_lock = asyncio.Lock()
        self._client = httpx.AsyncClient(
            base_url=config.base_url,
            headers=build_headers(config.user_agent),
            cookies=build_cookies(config.cf_clearance),
            timeout=config.timeout,
            proxy=config.proxy,
            follow_redirects=True,
        )

    async def get(self, path: str) -> str:
        """Fetch ``path`` and return the HTML body, or raise a typed transport error."""
        retrying = build_async_retrying(self._config)
        attempt = 0

        async def _attempt() -> str:
            nonlocal attempt
            attempt += 1
            await self._throttle()  # space every HTTP attempt, including retries
            try:
                response = await self._client.get(path)
            except httpx.TransportError as exc:
                logger.debug("GET %s failed on attempt %d: %r", path, attempt, exc)
                raise TransientTransportError from exc
            logger.debug("GET %s -> %s (attempt %d)", path, response.status_code, attempt)
            classify_response(response.status_code, response.headers, response.text)
            return response.text

        result: str = await retrying(_attempt)
        return result

    def set_cf_clearance(self, token: str) -> None:
        """Update the cf_clearance cookie used by subsequent requests."""
        self._config.cf_clearance = token
        self._client.cookies.set(CF_COOKIE_NAME, token)

    def set_user_agent(self, user_agent: str) -> None:
        """Update the User-Agent header used by subsequent requests."""
        self._config.user_agent = user_agent
        self._client.headers["User-Agent"] = user_agent

    async def aclose(self) -> None:
        await self._client.aclose()

    @property
    def is_closed(self) -> bool:
        return self._client.is_closed

    async def __aenter__(self) -> AsyncTransport:
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        await self.aclose()

    async def _throttle(self) -> None:
        interval = self._config.min_interval
        if interval <= 0:
            return
        # Serialize the gate so concurrent get() coroutines honor min_interval instead of
        # racing on self._last; each waits its turn behind the previous request.
        async with self._throttle_lock:
            wait = throttle_wait(self._last, time.monotonic(), interval)
            if wait > 0:
                await asyncio.sleep(wait)
            self._last = time.monotonic()

get(path) async

Fetch path and return the HTML body, or raise a typed transport error.

Source code in src/vgmdb_client/transport/async_client.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
async def get(self, path: str) -> str:
    """Fetch ``path`` and return the HTML body, or raise a typed transport error."""
    retrying = build_async_retrying(self._config)
    attempt = 0

    async def _attempt() -> str:
        nonlocal attempt
        attempt += 1
        await self._throttle()  # space every HTTP attempt, including retries
        try:
            response = await self._client.get(path)
        except httpx.TransportError as exc:
            logger.debug("GET %s failed on attempt %d: %r", path, attempt, exc)
            raise TransientTransportError from exc
        logger.debug("GET %s -> %s (attempt %d)", path, response.status_code, attempt)
        classify_response(response.status_code, response.headers, response.text)
        return response.text

    result: str = await retrying(_attempt)
    return result

set_cf_clearance(token)

Update the cf_clearance cookie used by subsequent requests.

Source code in src/vgmdb_client/transport/async_client.py 
67
68
69
70
def set_cf_clearance(self, token: str) -> None:
    """Update the cf_clearance cookie used by subsequent requests."""
    self._config.cf_clearance = token
    self._client.cookies.set(CF_COOKIE_NAME, token)

set_user_agent(user_agent)

Update the User-Agent header used by subsequent requests.

Source code in src/vgmdb_client/transport/async_client.py 
72
73
74
75
def set_user_agent(self, user_agent: str) -> None:
    """Update the User-Agent header used by subsequent requests."""
    self._config.user_agent = user_agent
    self._client.headers["User-Agent"] = user_agent

CloudflareChallengeError

Bases: TransportError

Raised when vgmdb returns a Cloudflare challenge.

This is not retried: a missing or stale cf_clearance token will not resolve by retrying. Supply or refresh the token and try again.

Source code in src/vgmdb_client/transport/errors.py 
15
16
17
18
19
20
21
22
23
24
25
26
class CloudflareChallengeError(TransportError):
    """Raised when vgmdb returns a Cloudflare challenge.

    This is not retried: a missing or stale ``cf_clearance`` token will not
    resolve by retrying. Supply or refresh the token and try again.
    """

    def __init__(
        self,
        message: str = "Cloudflare challenge detected; supply or refresh the cf_clearance token.",
    ) -> None:
        super().__init__(message)

NotFoundError

Bases: TransportError

Raised for an application-level 404 (no such resource).

Source code in src/vgmdb_client/transport/errors.py 
29
30
31
32
33
class NotFoundError(TransportError):
    """Raised for an application-level 404 (no such resource)."""

    def __init__(self, message: str = "The requested vgmdb resource was not found (404).") -> None:
        super().__init__(message)

RateLimitedError

Bases: TransportError

Raised for a 429 response.

Exposes retry_after (seconds) when the server provided a Retry-After header. The transport does not auto-retry rate limits.

Source code in src/vgmdb_client/transport/errors.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class RateLimitedError(TransportError):
    """Raised for a 429 response.

    Exposes ``retry_after`` (seconds) when the server provided a
    ``Retry-After`` header. The transport does not auto-retry rate limits.
    """

    def __init__(
        self,
        message: str = "vgmdb rate-limited the request (429).",
        *,
        retry_after: float | None = None,
    ) -> None:
        super().__init__(message)
        self.retry_after = retry_after

SyncTransport

Authenticated, Cloudflare-aware synchronous fetcher for vgmdb pages.

Source code in src/vgmdb_client/transport/sync_client.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class SyncTransport:
    """Authenticated, Cloudflare-aware synchronous fetcher for vgmdb pages."""

    def __init__(self, config: TransportConfig) -> None:
        self._config = config
        self._last: float | None = None
        self._client = httpx.Client(
            base_url=config.base_url,
            headers=build_headers(config.user_agent),
            cookies=build_cookies(config.cf_clearance),
            timeout=config.timeout,
            proxy=config.proxy,
            follow_redirects=True,
        )

    def get(self, path: str) -> str:
        """Fetch ``path`` and return the HTML body, or raise a typed transport error."""
        retrying = build_retrying(self._config)
        attempt = 0

        def _attempt() -> str:
            nonlocal attempt
            attempt += 1
            self._throttle()  # space every HTTP attempt, including retries
            try:
                response = self._client.get(path)
            except httpx.TransportError as exc:
                logger.debug("GET %s failed on attempt %d: %r", path, attempt, exc)
                raise TransientTransportError from exc
            logger.debug("GET %s -> %s (attempt %d)", path, response.status_code, attempt)
            classify_response(response.status_code, response.headers, response.text)
            return response.text

        result: str = retrying(_attempt)
        return result

    def set_cf_clearance(self, token: str) -> None:
        """Update the cf_clearance cookie used by subsequent requests."""
        self._config.cf_clearance = token
        self._client.cookies.set(CF_COOKIE_NAME, token)

    def set_user_agent(self, user_agent: str) -> None:
        """Update the User-Agent header used by subsequent requests."""
        self._config.user_agent = user_agent
        self._client.headers["User-Agent"] = user_agent

    def close(self) -> None:
        self._client.close()

    @property
    def is_closed(self) -> bool:
        return self._client.is_closed

    def __enter__(self) -> SyncTransport:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        self.close()

    def _throttle(self) -> None:
        interval = self._config.min_interval
        if interval <= 0:
            return
        wait = throttle_wait(self._last, time.monotonic(), interval)
        if wait > 0:
            time.sleep(wait)
        self._last = time.monotonic()

get(path)

Fetch path and return the HTML body, or raise a typed transport error.

Source code in src/vgmdb_client/transport/sync_client.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def get(self, path: str) -> str:
    """Fetch ``path`` and return the HTML body, or raise a typed transport error."""
    retrying = build_retrying(self._config)
    attempt = 0

    def _attempt() -> str:
        nonlocal attempt
        attempt += 1
        self._throttle()  # space every HTTP attempt, including retries
        try:
            response = self._client.get(path)
        except httpx.TransportError as exc:
            logger.debug("GET %s failed on attempt %d: %r", path, attempt, exc)
            raise TransientTransportError from exc
        logger.debug("GET %s -> %s (attempt %d)", path, response.status_code, attempt)
        classify_response(response.status_code, response.headers, response.text)
        return response.text

    result: str = retrying(_attempt)
    return result

set_cf_clearance(token)

Update the cf_clearance cookie used by subsequent requests.

Source code in src/vgmdb_client/transport/sync_client.py 
61
62
63
64
def set_cf_clearance(self, token: str) -> None:
    """Update the cf_clearance cookie used by subsequent requests."""
    self._config.cf_clearance = token
    self._client.cookies.set(CF_COOKIE_NAME, token)

set_user_agent(user_agent)

Update the User-Agent header used by subsequent requests.

Source code in src/vgmdb_client/transport/sync_client.py 
66
67
68
69
def set_user_agent(self, user_agent: str) -> None:
    """Update the User-Agent header used by subsequent requests."""
    self._config.user_agent = user_agent
    self._client.headers["User-Agent"] = user_agent

TransientTransportError

Bases: TransportError

Raised for retryable failures (connection errors, timeouts, 5xx).

Propagates only after retries are exhausted.

Source code in src/vgmdb_client/transport/errors.py 
53
54
55
56
57
58
59
60
class TransientTransportError(TransportError):
    """Raised for retryable failures (connection errors, timeouts, 5xx).

    Propagates only after retries are exhausted.
    """

    def __init__(self, message: str = "A transient transport error occurred.") -> None:
        super().__init__(message)

TransportConfig

Bases: BaseModel

Settings for a vgmdb transport client.

user_agent is required and must match the browser the cf_clearance token was issued for. A min_interval of 0 disables the politeness throttle.

Source code in src/vgmdb_client/transport/config.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class TransportConfig(BaseModel):
    """Settings for a vgmdb transport client.

    ``user_agent`` is required and must match the browser the ``cf_clearance``
    token was issued for. A ``min_interval`` of ``0`` disables the politeness
    throttle.
    """

    base_url: str = DEFAULT_BASE_URL
    user_agent: str
    cf_clearance: str | None = None
    timeout: float = Field(default=10.0, gt=0)
    max_retries: int = Field(default=3, ge=0)
    backoff_base: float = Field(default=0.5, gt=0)
    backoff_max: float = Field(default=8.0, gt=0)
    min_interval: float = Field(default=1.0, ge=0)
    proxy: str | None = None

TransportError

Bases: VgmdbClientError

Base class for all transport-layer failures.

Source code in src/vgmdb_client/transport/errors.py 
 8
 9
10
11
12
class TransportError(VgmdbClientError):
    """Base class for all transport-layer failures."""

    def __init__(self, message: str = "A transport error occurred.") -> None:
        super().__init__(message)