Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
106 changes: 58 additions & 48 deletions src/urlscan/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -370,57 +370,67 @@ def get_text(self, path: str, params: QueryParamTypes | None = None) -> str:
res = self._get(path, params=params)
return self._response_to_str(res)

def _get_error(self, res: ClientResponse) -> APIError | None:
try:
res.raise_for_status()
except httpx.HTTPStatusError as exc:
data: dict = exc.response.json()
message: str = data["message"]
description: str | None = data.get("description")
code: str | None = data.get("code")
type_: str | None = data.get("type")
# fallback to HTTP status code if "status" is missing
status: int = data.get("status") or exc.response.status_code

# ref. https://urlscan.io/docs/api/#ratelimit
if status == 429:
rate_limit_reset_after = float(
exc.response.headers.get("X-Rate-Limit-Reset-After", 0)
)
return RateLimitError(
message,
description=description,
status=status,
rate_limit_reset_after=rate_limit_reset_after,
)

def mapper(d: dict) -> ItemError:
title: str = d["title"]
status: int = d["status"]
code: str | None = d.get("code")
description: str | None = d.get("description")
detail: str | None = d.get("detail")
return ItemError(
title=title,
description=description,
detail=detail,
status=status,
code=code,
)

errors: list[ItemError] | None = None
if "errors" in data:
errors = [mapper(item) for item in data["errors"]]

return APIError(
def _map_http_status_error(self, exc: httpx.HTTPStatusError) -> APIError:
data: dict = exc.response.json()
message: str = data["message"]
description: str | None = data.get("description")
code: str | None = data.get("code")
type_: str | None = data.get("type")
# fallback to HTTP status code if "status" is missing
status: int = data.get("status") or exc.response.status_code

# ref. https://urlscan.io/docs/api/#ratelimit
if status == 429:
rate_limit_reset_after = float(
exc.response.headers.get("X-Rate-Limit-Reset-After", 0)
)
return RateLimitError(
message,
description=description,
status=status,
rate_limit_reset_after=rate_limit_reset_after,
)

def mapper(d: dict) -> ItemError:
title: str = d["title"]
status: int = d["status"]
code: str | None = d.get("code")
description: str | None = d.get("description")
detail: str | None = d.get("detail")
return ItemError(
title=title,
description=description,
detail=detail,
status=status,
code=code,
type_=type_,
errors=errors,
)

errors: list[ItemError] | None = None
if "errors" in data:
errors = [mapper(item) for item in data["errors"]]

return APIError(
message,
description=description,
status=status,
code=code,
type_=type_,
errors=errors,
)

def _get_error(self, res: ClientResponse) -> APIError | None:
try:
res.raise_for_status()
except httpx.HTTPStatusError as exc:
try:
return self._map_http_status_error(exc)
except (json.JSONDecodeError, UnicodeDecodeError):
# when error response is not JSON
return APIError(
message=exc.response.text,
status=exc.response.status_code,
)

return None

def _response_to_json(self, res: ClientResponse) -> dict:
Expand Down Expand Up @@ -476,9 +486,9 @@ def get_screenshot(self, uuid: str) -> BytesIO:
https://urlscan.io/docs/api/#screenshot

"""
res = self._get(f"/screenshots/{uuid}.png")
bio = BytesIO(res.content)
bio.name = res.basename
res = self.get_content(f"/screenshots/{uuid}.png")
bio = BytesIO(res)
bio.name = f"{uuid}.png"
return bio

def get_dom(self, uuid: str) -> str:
Expand Down
21 changes: 21 additions & 0 deletions tests/unit/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -532,6 +532,27 @@ def test_error_3(client: Client, httpserver: HTTPServer):
assert exc.errors is not None


def test_error_4(client: Client, httpserver: HTTPServer):
# non JSON error
httpserver.expect_request(
"/error",
method="GET",
).respond_with_data(
"Internal Server Error",
status=500,
content_type="text/plain",
)
with pytest.raises(APIError) as exc_info:
client.get_json("/error")

exc = exc_info.value
assert exc.status == 500
assert exc.message == "Internal Server Error"
assert exc.code is None
assert exc.type is None
assert exc.errors is None


def test_get_response(client: Client, httpserver: HTTPServer):
httpserver.expect_request("/responses/dummy/", method="GET").respond_with_data(
"dummy", content_type="text/plain"
Expand Down