Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
File size: 3,415 Bytes
ee0b555 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 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 97 98 99 100 101 | """Credential-ready anonymous HTTP transport for the public catalog API."""
from __future__ import annotations
import json
import os
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from typing import Any
from pwc_cli import __version__
DEFAULT_API_URL = "https://paperswithcode.co/api/v1"
MAX_RESPONSE_BYTES = 2 * 1024 * 1024
class TransportError(RuntimeError):
pass
class HTTPStatusError(TransportError):
def __init__(self, status: int, detail: str):
super().__init__(f"API request failed ({status}): {detail}")
self.status = status
self.detail = detail
class ResponseError(RuntimeError):
pass
@dataclass(frozen=True)
class Response:
body: bytes
headers: dict[str, str]
def json(self) -> Any:
try:
return json.loads(self.body)
except (UnicodeDecodeError, json.JSONDecodeError) as error:
raise ResponseError("API returned invalid JSON") from error
class Client:
def __init__(
self,
base_url: str | None = None,
credential: str | None = None,
timeout: float = 30,
):
configured = base_url or os.environ.get("PWC_API_URL") or DEFAULT_API_URL
self.base_url = configured.rstrip("/")
self.credential = credential
self.timeout = timeout
def get(
self, path: str, params: dict[str, object | None] | None = None
) -> Response:
query = urllib.parse.urlencode(
{
key: str(value).lower() if isinstance(value, bool) else value
for key, value in (params or {}).items()
if value is not None
},
doseq=True,
)
url = f"{self.base_url}/{path.lstrip('/')}"
if query:
url += f"?{query}"
headers = {
"Accept": "application/json, text/plain;q=0.9",
"User-Agent": f"pwc-cli/{__version__}",
}
if self.credential:
headers["Authorization"] = f"Bearer {self.credential}"
request = urllib.request.Request(url, headers=headers)
try:
with urllib.request.urlopen(request, timeout=self.timeout) as response:
length = response.headers.get("Content-Length")
if length and int(length) > MAX_RESPONSE_BYTES:
raise ResponseError("API response exceeded the client byte limit")
body = response.read(MAX_RESPONSE_BYTES + 1)
if len(body) > MAX_RESPONSE_BYTES:
raise ResponseError("API response exceeded the client byte limit")
return Response(
body,
{key.lower(): value for key, value in response.headers.items()},
)
except urllib.error.HTTPError as error:
detail = error.read(4096).decode("utf-8", errors="replace")
raise HTTPStatusError(error.code, detail) from error
except urllib.error.URLError as error:
raise TransportError(f"API request failed: {error.reason}") from error
except TimeoutError as error:
# Read timeouts surface as a bare TimeoutError, not a URLError.
raise TransportError("API request timed out") from error
except OSError as error:
raise TransportError(f"API request failed: {error}") from error
|