Client

urlscan.Client

Source code in src/urlscan/client.py

class Client:
    def __init__(
        self,
        api_key: str,
        base_url: str = BASE_URL,
        user_agent: str = USER_AGENT,
        trust_env: bool = False,
        timeout: TimeoutTypes = 60,
        proxy: str | None = None,
        verify: bool = True,
        retry: bool = False,
    ):
        """
        Args:
            api_key (str): Your urlscan.io API key.
            base_url (str, optional): Base URL. Defaults to BASE_URL.
            user_agent (str, optional): User agent. Defaults to USER_AGENT.
            trust_env (bool, optional): Enable or disable usage of environment variables for configuration. Defaults to False.
            timeout (TimeoutTypes, optional): timeout configuration to use when sending request. Defaults to 60.
            proxy (str | None, optional): Proxy URL where all the traffic should be routed. Defaults to None.
            verify (bool, optional): Either `True` to use an SSL context with the default CA bundle, `False` to disable verification. Defaults to True.
            retry (bool, optional): Whether to use automatic X-Rate-Limit-Reset-After HTTP header based retry. Defaults to False.
        """
        self._api_key = api_key
        self._base_url = base_url
        self._user_agent = user_agent
        self._trust_env = trust_env
        self._timeout = timeout
        self._proxy = proxy
        self._verify = verify
        self._retry = retry

        self._session: httpx.Client | None = None
        self._rate_limit_memo: RateLimitMemo = {
            "public": None,
            "private": None,
            "unlisted": None,
            "retrieve": None,
            "search": None,
        }

        self._scan_uuid_timestamp_memo: dict[str, float] = {}

    def __enter__(self):
        return self

    def __exit__(self, item_type: Any, value: Any, traceback: Any):
        self.close()

    def close(self):
        if self._session:
            self._session.close()
            self._session = None

    def _get_session(self) -> httpx.Client:
        if self._session:
            return self._session

        headers = _compact(
            {
                "User-Agent": self._user_agent,
                "API-Key": self._api_key,
            }
        )
        transport: httpx.HTTPTransport | None = None
        if self._retry:
            transport = RetryTransport()

        self._session = httpx.Client(
            base_url=self._base_url,
            headers=headers,
            timeout=self._timeout,
            proxy=self._proxy,
            verify=self._verify,
            trust_env=self._trust_env,
            transport=transport,
        )
        return self._session

    def _get_action(self, request: httpx.Request) -> ActionType | None:
        path = request.url.path
        if request.method == "GET":
            if path == "/api/v1/search/":
                return "search"

            if path.startswith("/api/v1/result/"):
                return "retrieve"

            return None

        if request.method == "POST":
            if path != "/api/v1/scan/":
                return None

            if request.headers.get("Content-Type") != "application/json":
                return None

            with contextlib.suppress(json.JSONDecodeError):
                data: dict = json.loads(request.content)
                return data.get("visibility")

        return None

    def _send_request(
        self, session: httpx.Client, request: httpx.Request
    ) -> ClientResponse:
        # let it automatic retry if retry is enabled
        if self._retry:
            return ClientResponse(session.send(request))

        action = self._get_action(request)
        if action:
            rate_limit: RateLimit | None = self._rate_limit_memo.get(action)
            if rate_limit:
                utcnow = datetime.datetime.now(datetime.timezone.utc)
                if rate_limit.remaining == 0 and rate_limit.reset > utcnow:
                    raise RateLimitRemainingError(
                        f"{action} is rate limited. Wait until {utcnow}."
                    )

        res = ClientResponse(session.send(request))

        # use action in response headers
        action = res.headers.get("X-Rate-Limit-Action")
        if action:
            remaining = res.headers.get("X-Rate-Limit-Remaining")
            reset = res.headers.get("X-Rate-Limit-Reset")
            if remaining and reset:
                self._rate_limit_memo[action] = RateLimit(
                    remaining=int(remaining),
                    reset=parse_datetime(reset),
                )

        return res

    def get(self, path: str, params: QueryParamTypes | None = None) -> ClientResponse:
        """Send a GET request to a given API endpoint.

        Args:
            path (str): Path to API endpoint.
            params (QueryParamTypes | None, optional): Query parameters. Defaults to None.

        Returns:
            ClientResponse: Response.
        """
        session = self._get_session()
        req = session.build_request("GET", path, params=params)
        return self._send_request(session, req)

    def get_json(self, path: str, params: QueryParamTypes | None = None) -> dict:
        res = self.get(path, params=params)
        return self._response_to_json(res)

    def post(
        self,
        path: str,
        json: Any | None = None,
        data: RequestData | None = None,
    ) -> ClientResponse:
        """Send a POST request to a given API endpoint.

        Args:
            path (str): Path.
            json (Any | None, optional): Dict to send in request body as JSON. Defaults to None.
            data (RequestData | None, optional): Dict to send in request body. Defaults to None.

        Returns:
            ClientResponse: Response.
        """
        session = self._get_session()
        req = session.build_request("POST", path, json=json, data=data)
        return self._send_request(session, req)

    def download(
        self,
        path: str,
        file: BinaryIO,
        params: QueryParamTypes | None = None,
    ) -> None:
        """Download a file from a given API endpoint.

        Args:
            path (str): Path to API endpoint.
            file (BinaryIO): File object to write to.
            params (QueryParamTypes | None, optional): Query parameters. Defaults to None.

        Returns:
            BytesIO: File content.
        """
        res = self.get(path, params=params)
        file.write(res.content)
        return

    def get_content(self, path: str, params: QueryParamTypes | None = None) -> bytes:
        res = self.get(path, params=params)
        return self._response_to_content(res)

    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_result(self, uuid: str) -> dict:
        """Get a result of a scan by UUID.

        Args:
            uuid (str): UUID.

        Returns:
            Dict: Scan result.

        Reference:
            https://urlscan.io/docs/api/#result
        """
        return self.get_json(f"/api/v1/result/{uuid}/")

    def get_screenshot(self, uuid: str) -> BytesIO:
        """Get a screenshot of a scan by UUID.

        Args:
            uuid (str): UUID.

        Returns:
            : Screenshot (img/png) as bytes.

        Reference:
            https://urlscan.io/docs/api/#screenshot
        """
        res = self.get(f"/screenshots/{uuid}.png")
        bio = BytesIO(res.content)
        bio.name = res.basename
        return bio

    def get_dom(self, uuid: str) -> str:
        """Get a DOM of a scan by UUID.

        Args:
            uuid (str): UUID

        Returns:
            str: DOM as a string.

        Reference:
            https://urlscan.io/docs/api/#dom
        """
        return self.get_text(f"/dom/{uuid}/")

    def search(
        self,
        q: str = "",
        size: int = 100,
        limit: int | None = None,
        search_after: str | None = None,
    ) -> SearchIterator:
        """Search.

        Args:
            q (str): Query term. Defaults to "".
            size (int, optional): Number of results returned in a search. Defaults to 100.
            limit (int | None, optional): . Defaults to None.
            search_after (str | None, optional): Maximum number of results that will be returned by the iterator. Defaults to None.

        Returns:
            SearchIterator: Search iterator.

        Reference:
            https://urlscan.io/docs/api/#search
        """
        return SearchIterator(
            self,
            q=q,
            size=size,
            limit=limit,
            search_after=search_after,
        )

    def scan(
        self,
        url: str,
        *,
        visibility: VisibilityType,
        tags: list[str] | None = None,
        customagent: str | None = None,
        referer: str | None = None,
        override_safety: Any = None,
        country: str | None = None,
    ) -> dict:
        """Scan a given URL.

        Args:
            url (str): URL to scan.
            visibility (VisibilityType): Visibility of the scan. Can be "public", "private", or "unlisted".
            tags (list[str] | None, optional): Tags to be attached. Defaults to None.
            customagent (str | None, optional): Custom user agent. Defaults to None.
            referer (str | None, optional): Referer. Defaults to None.
            override_safety (Any, optional): If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.
            country (str | None, optional): Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.

        Returns:
            dict: Scan response.

        Reference:
            https://urlscan.io/docs/api/#scan
        """
        data = _compact(
            {
                "url": url,
                "tags": tags,
                "visibility": visibility,
                "customagent": customagent,
                "referer": referer,
                "overrideSafety": override_safety,
                "country": country,
            }
        )
        res = self.post("/api/v1/scan/", json=data)
        json_res = self._response_to_json(res)

        json_visibility = json_res.get("visibility")
        if json_visibility is not None and json_visibility != visibility:
            logger.warning(f"Visibility is enforced to {json_visibility}.")

        # memoize the scan UUID & timestamp
        uuid = json_res.get("uuid")
        if isinstance(uuid, str):
            self._scan_uuid_timestamp_memo[uuid] = time.time()

        return json_res

    def bulk_scan(
        self,
        urls: list[str],
        *,
        visibility: VisibilityType,
        tags: list[str] | None = None,
        customagent: str | None = None,
        referer: str | None = None,
        override_safety: Any = None,
        country: str | None = None,
    ) -> list[tuple[str, dict | Exception]]:
        """Scan multiple URLs in bulk.

        Args:
            urls (list[str]): List of URLs to scan.
            visibility (VisibilityType): Visibility of the scan. Can be "public", "private", or "unlisted".
            tags (list[str] | None, optional): Tags to be attached. Defaults to None.
            customagent (str | None, optional): Custom user agent. Defaults to None.
            referer (str | None, optional): Referer. Defaults to None.
            override_safety (Any, optional): If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.
            country (str | None, optional): Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.

        Returns:
            list[tuple[str, dict | Exception]]: A list of tuples of (url, scan response or error).

        Reference:
            https://urlscan.io/docs/api/#scan
        """

        def inner(url: str) -> dict | Exception:
            try:
                return self.scan(
                    url,
                    visibility=visibility,
                    tags=tags,
                    customagent=customagent,
                    referer=referer,
                    override_safety=override_safety,
                    country=country,
                )
            except Exception as e:
                return e

        return [(url, inner(url)) for url in urls]

    def wait_for_result(
        self,
        uuid: str,
        timeout: float = 60.0,
        interval: float = 1.0,
        initial_wait: float | None = 10.0,
    ) -> None:
        """Wait for a scan result to be available.

        Args:
            uuid (str): UUID of a result.
            timeout (float, optional): Timeout in seconds. Defaults to 60.0.
            interval (float, optional): Interval in seconds. Defaults to 1.0.
            initial_wait (float | None, optional): Initial wait time in seconds. Set None to disable. Defaults to 10.0.
        """
        session = self._get_session()
        req = session.build_request("HEAD", f"/api/v1/result/{uuid}/")

        scanned_at = self._scan_uuid_timestamp_memo.get(uuid)
        if scanned_at and initial_wait:
            elapsed = time.time() - scanned_at
            if elapsed < initial_wait:
                time.sleep(initial_wait - elapsed)

        start_time = time.time()
        while True:
            res = self._send_request(session, req)
            if res.status_code == 200:
                self._scan_uuid_timestamp_memo.pop(uuid, None)
                return

            if time.time() - start_time > timeout:
                raise TimeoutError("Timeout waiting for scan result.")

            time.sleep(interval)

    def scan_and_get_result(
        self,
        url: str,
        visibility: VisibilityType,
        tags: list[str] | None = None,
        customagent: str | None = None,
        referer: str | None = None,
        override_safety: Any = None,
        country: str | None = None,
        timeout: float = 60.0,
        interval: float = 1.0,
        initial_wait: float | None = 10.0,
    ):
        """Scan a given URL, wait for a result and get it.

        Args:
            url (str): URL to scan.
            visibility (VisibilityType): Visibility of the scan. Can be "public", "private", or "unlisted".
            tags (list[str] | None, optional): Tags to be attached. Defaults to None.
            customagent (str | None, optional): Custom user agent. Defaults to None.
            referer (str | None, optional): Referer. Defaults to None.
            override_safety (Any, optional): If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.
            country (str | None, optional): Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.
            timeout (float, optional): Timeout for waiting a result in seconds. Defaults to 60.0.
            interval (float, optional): Interval in seconds. Defaults to 1.0.
            initial_wait (float | None, optional): Initial wait time in seconds. Set None to disable. Defaults to 10.0.

        Returns:
            dict: Scan result.

        Reference:
            https://urlscan.io/docs/api/#scan
        """
        res = self.scan(
            url,
            visibility=visibility,
            tags=tags,
            customagent=customagent,
            referer=referer,
            override_safety=override_safety,
            country=country,
        )
        uuid: str = res["uuid"]
        self.wait_for_result(
            uuid, timeout=timeout, interval=interval, initial_wait=initial_wait
        )
        return self.get_result(uuid)

    def bulk_scan_and_get_results(
        self,
        urls: list[str],
        visibility: VisibilityType,
        tags: list[str] | None = None,
        customagent: str | None = None,
        referer: str | None = None,
        override_safety: Any = None,
        country: str | None = None,
        timeout: float = 60.0,
        interval: float = 1.0,
        initial_wait: float | None = 10.0,
    ) -> list[tuple[str, dict | Exception]]:
        """Scan URLs, wait for results and get them.

        Args:
            urls (list[str]): URLs to scan.
            visibility (VisibilityType): Visibility of the scan. Can be "public", "private", or "unlisted".
            tags (list[str] | None, optional): Tags to be attached. Defaults to None.
            customagent (str | None, optional): Custom user agent. Defaults to None.
            referer (str | None, optional): Referer. Defaults to None.
            override_safety (Any, optional): If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.
            country (str | None, optional): Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.
            timeout (float, optional): Timeout for waiting a result in seconds. Defaults to 60.0.
            interval (float, optional): Interval in seconds. Defaults to 1.0.
            initial_wait (float | None, optional): Initial wait time in seconds. Set None to disable. Defaults to 10.0.

        Returns:
            list[tuple[str, dict | Exception]]: A list of tuples of (url, result or error).

        Reference:
            https://urlscan.io/docs/api/#scan
        """

        responses = self.bulk_scan(
            urls,
            visibility=visibility,
            tags=tags,
            customagent=customagent,
            referer=referer,
            override_safety=override_safety,
            country=country,
        )

        def mapping(res_or_error: dict | Exception) -> dict | Exception:
            if isinstance(res_or_error, Exception):
                return res_or_error

            uuid: str = res_or_error["uuid"]
            self.wait_for_result(
                uuid, timeout=timeout, interval=interval, initial_wait=initial_wait
            )
            return self.get_result(uuid)

        return [(url, mapping(res_or_error)) for url, res_or_error in responses]

    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")
            status: int = data["status"]

            # 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,
                )

            return APIError(message, description=description, status=status)

        return None

    def _response_to_json(self, res: ClientResponse) -> dict:
        error = self._get_error(res)
        if error:
            raise error

        return res.json()

    def _response_to_str(self, res: ClientResponse) -> str:
        error = self._get_error(res)
        if error:
            raise error

        return res.text

    def _response_to_content(self, res: ClientResponse) -> bytes:
        error = self._get_error(res)
        if error:
            raise error

        return res.content

__init__(api_key, base_url=BASE_URL, user_agent=USER_AGENT, trust_env=False, timeout=60, proxy=None, verify=True, retry=False)

Parameters:

Name	Type	Description	Default
api_key	str	Your urlscan.io API key.	required
base_url	str	Base URL. Defaults to BASE_URL.	BASE_URL
user_agent	str	User agent. Defaults to USER_AGENT.	USER_AGENT
trust_env	bool	Enable or disable usage of environment variables for configuration. Defaults to False.	False
timeout	TimeoutTypes	timeout configuration to use when sending request. Defaults to 60.	60
proxy	str | None	Proxy URL where all the traffic should be routed. Defaults to None.	None
verify	bool	Either True to use an SSL context with the default CA bundle, False to disable verification. Defaults to True.	True
retry	bool	Whether to use automatic X-Rate-Limit-Reset-After HTTP header based retry. Defaults to False.	False

Source code in src/urlscan/client.py

def __init__(
    self,
    api_key: str,
    base_url: str = BASE_URL,
    user_agent: str = USER_AGENT,
    trust_env: bool = False,
    timeout: TimeoutTypes = 60,
    proxy: str | None = None,
    verify: bool = True,
    retry: bool = False,
):
    """
    Args:
        api_key (str): Your urlscan.io API key.
        base_url (str, optional): Base URL. Defaults to BASE_URL.
        user_agent (str, optional): User agent. Defaults to USER_AGENT.
        trust_env (bool, optional): Enable or disable usage of environment variables for configuration. Defaults to False.
        timeout (TimeoutTypes, optional): timeout configuration to use when sending request. Defaults to 60.
        proxy (str | None, optional): Proxy URL where all the traffic should be routed. Defaults to None.
        verify (bool, optional): Either `True` to use an SSL context with the default CA bundle, `False` to disable verification. Defaults to True.
        retry (bool, optional): Whether to use automatic X-Rate-Limit-Reset-After HTTP header based retry. Defaults to False.
    """
    self._api_key = api_key
    self._base_url = base_url
    self._user_agent = user_agent
    self._trust_env = trust_env
    self._timeout = timeout
    self._proxy = proxy
    self._verify = verify
    self._retry = retry

    self._session: httpx.Client | None = None
    self._rate_limit_memo: RateLimitMemo = {
        "public": None,
        "private": None,
        "unlisted": None,
        "retrieve": None,
        "search": None,
    }

    self._scan_uuid_timestamp_memo: dict[str, float] = {}

bulk_scan(urls, *, visibility, tags=None, customagent=None, referer=None, override_safety=None, country=None)

Scan multiple URLs in bulk.

Parameters:

Name	Type	Description	Default
urls	list[str]	List of URLs to scan.	required
visibility	VisibilityType	Visibility of the scan. Can be "public", "private", or "unlisted".	required
tags	list[str] | None	Tags to be attached. Defaults to None.	None
customagent	str | None	Custom user agent. Defaults to None.	None
referer	str | None	Referer. Defaults to None.	None
override_safety	Any	If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.	None
country	str | None	Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.	None

Returns:

Type	Description
list[tuple[str, dict | Exception]]	list[tuple[str, dict | Exception]]: A list of tuples of (url, scan response or error).

Reference

https://urlscan.io/docs/api/#scan

Source code in src/urlscan/client.py

def bulk_scan(
    self,
    urls: list[str],
    *,
    visibility: VisibilityType,
    tags: list[str] | None = None,
    customagent: str | None = None,
    referer: str | None = None,
    override_safety: Any = None,
    country: str | None = None,
) -> list[tuple[str, dict | Exception]]:
    """Scan multiple URLs in bulk.

    Args:
        urls (list[str]): List of URLs to scan.
        visibility (VisibilityType): Visibility of the scan. Can be "public", "private", or "unlisted".
        tags (list[str] | None, optional): Tags to be attached. Defaults to None.
        customagent (str | None, optional): Custom user agent. Defaults to None.
        referer (str | None, optional): Referer. Defaults to None.
        override_safety (Any, optional): If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.
        country (str | None, optional): Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.

    Returns:
        list[tuple[str, dict | Exception]]: A list of tuples of (url, scan response or error).

    Reference:
        https://urlscan.io/docs/api/#scan
    """

    def inner(url: str) -> dict | Exception:
        try:
            return self.scan(
                url,
                visibility=visibility,
                tags=tags,
                customagent=customagent,
                referer=referer,
                override_safety=override_safety,
                country=country,
            )
        except Exception as e:
            return e

    return [(url, inner(url)) for url in urls]

bulk_scan_and_get_results(urls, visibility, tags=None, customagent=None, referer=None, override_safety=None, country=None, timeout=60.0, interval=1.0, initial_wait=10.0)

Scan URLs, wait for results and get them.

Parameters:

Name	Type	Description	Default
urls	list[str]	URLs to scan.	required
visibility	VisibilityType	Visibility of the scan. Can be "public", "private", or "unlisted".	required
tags	list[str] | None	Tags to be attached. Defaults to None.	None
customagent	str | None	Custom user agent. Defaults to None.	None
referer	str | None	Referer. Defaults to None.	None
override_safety	Any	If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.	None
country	str | None	Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.	None
timeout	float	Timeout for waiting a result in seconds. Defaults to 60.0.	60.0
interval	float	Interval in seconds. Defaults to 1.0.	1.0
initial_wait	float | None	Initial wait time in seconds. Set None to disable. Defaults to 10.0.	10.0

Returns:

Type	Description
list[tuple[str, dict | Exception]]	list[tuple[str, dict | Exception]]: A list of tuples of (url, result or error).

Reference

https://urlscan.io/docs/api/#scan

Source code in src/urlscan/client.py

def bulk_scan_and_get_results(
    self,
    urls: list[str],
    visibility: VisibilityType,
    tags: list[str] | None = None,
    customagent: str | None = None,
    referer: str | None = None,
    override_safety: Any = None,
    country: str | None = None,
    timeout: float = 60.0,
    interval: float = 1.0,
    initial_wait: float | None = 10.0,
) -> list[tuple[str, dict | Exception]]:
    """Scan URLs, wait for results and get them.

    Args:
        urls (list[str]): URLs to scan.
        visibility (VisibilityType): Visibility of the scan. Can be "public", "private", or "unlisted".
        tags (list[str] | None, optional): Tags to be attached. Defaults to None.
        customagent (str | None, optional): Custom user agent. Defaults to None.
        referer (str | None, optional): Referer. Defaults to None.
        override_safety (Any, optional): If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.
        country (str | None, optional): Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.
        timeout (float, optional): Timeout for waiting a result in seconds. Defaults to 60.0.
        interval (float, optional): Interval in seconds. Defaults to 1.0.
        initial_wait (float | None, optional): Initial wait time in seconds. Set None to disable. Defaults to 10.0.

    Returns:
        list[tuple[str, dict | Exception]]: A list of tuples of (url, result or error).

    Reference:
        https://urlscan.io/docs/api/#scan
    """

    responses = self.bulk_scan(
        urls,
        visibility=visibility,
        tags=tags,
        customagent=customagent,
        referer=referer,
        override_safety=override_safety,
        country=country,
    )

    def mapping(res_or_error: dict | Exception) -> dict | Exception:
        if isinstance(res_or_error, Exception):
            return res_or_error

        uuid: str = res_or_error["uuid"]
        self.wait_for_result(
            uuid, timeout=timeout, interval=interval, initial_wait=initial_wait
        )
        return self.get_result(uuid)

    return [(url, mapping(res_or_error)) for url, res_or_error in responses]

download(path, file, params=None)

Download a file from a given API endpoint.

Parameters:

Name	Type	Description	Default
path	str	Path to API endpoint.	required
file	BinaryIO	File object to write to.	required
params	QueryParamTypes | None	Query parameters. Defaults to None.	None

Returns:

Name	Type	Description
BytesIO	None	File content.

Source code in src/urlscan/client.py

def download(
    self,
    path: str,
    file: BinaryIO,
    params: QueryParamTypes | None = None,
) -> None:
    """Download a file from a given API endpoint.

    Args:
        path (str): Path to API endpoint.
        file (BinaryIO): File object to write to.
        params (QueryParamTypes | None, optional): Query parameters. Defaults to None.

    Returns:
        BytesIO: File content.
    """
    res = self.get(path, params=params)
    file.write(res.content)
    return

get(path, params=None)

Send a GET request to a given API endpoint.

Parameters:

Name	Type	Description	Default
path	str	Path to API endpoint.	required
params	QueryParamTypes | None	Query parameters. Defaults to None.	None

Returns:

Name	Type	Description
ClientResponse	ClientResponse	Response.

Source code in src/urlscan/client.py

def get(self, path: str, params: QueryParamTypes | None = None) -> ClientResponse:
    """Send a GET request to a given API endpoint.

    Args:
        path (str): Path to API endpoint.
        params (QueryParamTypes | None, optional): Query parameters. Defaults to None.

    Returns:
        ClientResponse: Response.
    """
    session = self._get_session()
    req = session.build_request("GET", path, params=params)
    return self._send_request(session, req)

get_dom(uuid)

Get a DOM of a scan by UUID.

Parameters:

Name	Type	Description	Default
uuid	str	UUID	required

Returns:

Name	Type	Description
str	str	DOM as a string.

Reference

https://urlscan.io/docs/api/#dom

Source code in src/urlscan/client.py

def get_dom(self, uuid: str) -> str:
    """Get a DOM of a scan by UUID.

    Args:
        uuid (str): UUID

    Returns:
        str: DOM as a string.

    Reference:
        https://urlscan.io/docs/api/#dom
    """
    return self.get_text(f"/dom/{uuid}/")

get_result(uuid)

Get a result of a scan by UUID.

Parameters:

Name	Type	Description	Default
uuid	str	UUID.	required

Returns:

Name	Type	Description
Dict	dict	Scan result.

Reference

https://urlscan.io/docs/api/#result

Source code in src/urlscan/client.py

def get_result(self, uuid: str) -> dict:
    """Get a result of a scan by UUID.

    Args:
        uuid (str): UUID.

    Returns:
        Dict: Scan result.

    Reference:
        https://urlscan.io/docs/api/#result
    """
    return self.get_json(f"/api/v1/result/{uuid}/")

get_screenshot(uuid)

Get a screenshot of a scan by UUID.

Parameters:

Name	Type	Description	Default
uuid	str	UUID.	required

Returns:

Type	Description
BytesIO	Screenshot (img/png) as bytes.

Reference

https://urlscan.io/docs/api/#screenshot

Source code in src/urlscan/client.py

def get_screenshot(self, uuid: str) -> BytesIO:
    """Get a screenshot of a scan by UUID.

    Args:
        uuid (str): UUID.

    Returns:
        : Screenshot (img/png) as bytes.

    Reference:
        https://urlscan.io/docs/api/#screenshot
    """
    res = self.get(f"/screenshots/{uuid}.png")
    bio = BytesIO(res.content)
    bio.name = res.basename
    return bio

post(path, json=None, data=None)

Send a POST request to a given API endpoint.

Parameters:

Name	Type	Description	Default
path	str	Path.	required
json	Any | None	Dict to send in request body as JSON. Defaults to None.	None
data	RequestData | None	Dict to send in request body. Defaults to None.	None

Returns:

Name	Type	Description
ClientResponse	ClientResponse	Response.

Source code in src/urlscan/client.py

def post(
    self,
    path: str,
    json: Any | None = None,
    data: RequestData | None = None,
) -> ClientResponse:
    """Send a POST request to a given API endpoint.

    Args:
        path (str): Path.
        json (Any | None, optional): Dict to send in request body as JSON. Defaults to None.
        data (RequestData | None, optional): Dict to send in request body. Defaults to None.

    Returns:
        ClientResponse: Response.
    """
    session = self._get_session()
    req = session.build_request("POST", path, json=json, data=data)
    return self._send_request(session, req)

scan(url, *, visibility, tags=None, customagent=None, referer=None, override_safety=None, country=None)

Scan a given URL.

Parameters:

Name	Type	Description	Default
url	str	URL to scan.	required
visibility	VisibilityType	Visibility of the scan. Can be "public", "private", or "unlisted".	required
tags	list[str] | None	Tags to be attached. Defaults to None.	None
customagent	str | None	Custom user agent. Defaults to None.	None
referer	str | None	Referer. Defaults to None.	None
override_safety	Any	If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.	None
country	str | None	Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.	None

Returns:

Name	Type	Description
dict	dict	Scan response.

Reference

https://urlscan.io/docs/api/#scan

Source code in src/urlscan/client.py

def scan(
    self,
    url: str,
    *,
    visibility: VisibilityType,
    tags: list[str] | None = None,
    customagent: str | None = None,
    referer: str | None = None,
    override_safety: Any = None,
    country: str | None = None,
) -> dict:
    """Scan a given URL.

    Args:
        url (str): URL to scan.
        visibility (VisibilityType): Visibility of the scan. Can be "public", "private", or "unlisted".
        tags (list[str] | None, optional): Tags to be attached. Defaults to None.
        customagent (str | None, optional): Custom user agent. Defaults to None.
        referer (str | None, optional): Referer. Defaults to None.
        override_safety (Any, optional): If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.
        country (str | None, optional): Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.

    Returns:
        dict: Scan response.

    Reference:
        https://urlscan.io/docs/api/#scan
    """
    data = _compact(
        {
            "url": url,
            "tags": tags,
            "visibility": visibility,
            "customagent": customagent,
            "referer": referer,
            "overrideSafety": override_safety,
            "country": country,
        }
    )
    res = self.post("/api/v1/scan/", json=data)
    json_res = self._response_to_json(res)

    json_visibility = json_res.get("visibility")
    if json_visibility is not None and json_visibility != visibility:
        logger.warning(f"Visibility is enforced to {json_visibility}.")

    # memoize the scan UUID & timestamp
    uuid = json_res.get("uuid")
    if isinstance(uuid, str):
        self._scan_uuid_timestamp_memo[uuid] = time.time()

    return json_res

scan_and_get_result(url, visibility, tags=None, customagent=None, referer=None, override_safety=None, country=None, timeout=60.0, interval=1.0, initial_wait=10.0)

Scan a given URL, wait for a result and get it.

Parameters:

Name	Type	Description	Default
url	str	URL to scan.	required
visibility	VisibilityType	Visibility of the scan. Can be "public", "private", or "unlisted".	required
tags	list[str] | None	Tags to be attached. Defaults to None.	None
customagent	str | None	Custom user agent. Defaults to None.	None
referer	str | None	Referer. Defaults to None.	None
override_safety	Any	If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.	None
country	str | None	Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.	None
timeout	float	Timeout for waiting a result in seconds. Defaults to 60.0.	60.0
interval	float	Interval in seconds. Defaults to 1.0.	1.0
initial_wait	float | None	Initial wait time in seconds. Set None to disable. Defaults to 10.0.	10.0

Returns:

Name	Type	Description
dict		Scan result.

Reference

https://urlscan.io/docs/api/#scan

Source code in src/urlscan/client.py

def scan_and_get_result(
    self,
    url: str,
    visibility: VisibilityType,
    tags: list[str] | None = None,
    customagent: str | None = None,
    referer: str | None = None,
    override_safety: Any = None,
    country: str | None = None,
    timeout: float = 60.0,
    interval: float = 1.0,
    initial_wait: float | None = 10.0,
):
    """Scan a given URL, wait for a result and get it.

    Args:
        url (str): URL to scan.
        visibility (VisibilityType): Visibility of the scan. Can be "public", "private", or "unlisted".
        tags (list[str] | None, optional): Tags to be attached. Defaults to None.
        customagent (str | None, optional): Custom user agent. Defaults to None.
        referer (str | None, optional): Referer. Defaults to None.
        override_safety (Any, optional): If set to any value, this will disable reclassification of URLs with potential PII in them. Defaults to None.
        country (str | None, optional): Specify which country the scan should be performed from (2-Letter ISO-3166-1 alpha-2 country. Defaults to None.
        timeout (float, optional): Timeout for waiting a result in seconds. Defaults to 60.0.
        interval (float, optional): Interval in seconds. Defaults to 1.0.
        initial_wait (float | None, optional): Initial wait time in seconds. Set None to disable. Defaults to 10.0.

    Returns:
        dict: Scan result.

    Reference:
        https://urlscan.io/docs/api/#scan
    """
    res = self.scan(
        url,
        visibility=visibility,
        tags=tags,
        customagent=customagent,
        referer=referer,
        override_safety=override_safety,
        country=country,
    )
    uuid: str = res["uuid"]
    self.wait_for_result(
        uuid, timeout=timeout, interval=interval, initial_wait=initial_wait
    )
    return self.get_result(uuid)

search(q='', size=100, limit=None, search_after=None)

Search.

Parameters:

Name	Type	Description	Default
q	str	Query term. Defaults to "".	''
size	int	Number of results returned in a search. Defaults to 100.	100
limit	int | None	. Defaults to None.	None
search_after	str | None	Maximum number of results that will be returned by the iterator. Defaults to None.	None

Returns:

Name	Type	Description
SearchIterator	SearchIterator	Search iterator.

Reference

https://urlscan.io/docs/api/#search

Source code in src/urlscan/client.py

def search(
    self,
    q: str = "",
    size: int = 100,
    limit: int | None = None,
    search_after: str | None = None,
) -> SearchIterator:
    """Search.

    Args:
        q (str): Query term. Defaults to "".
        size (int, optional): Number of results returned in a search. Defaults to 100.
        limit (int | None, optional): . Defaults to None.
        search_after (str | None, optional): Maximum number of results that will be returned by the iterator. Defaults to None.

    Returns:
        SearchIterator: Search iterator.

    Reference:
        https://urlscan.io/docs/api/#search
    """
    return SearchIterator(
        self,
        q=q,
        size=size,
        limit=limit,
        search_after=search_after,
    )

wait_for_result(uuid, timeout=60.0, interval=1.0, initial_wait=10.0)

Wait for a scan result to be available.

Parameters:

Name	Type	Description	Default
uuid	str	UUID of a result.	required
timeout	float	Timeout in seconds. Defaults to 60.0.	60.0
interval	float	Interval in seconds. Defaults to 1.0.	1.0
initial_wait	float | None	Initial wait time in seconds. Set None to disable. Defaults to 10.0.	10.0

Source code in src/urlscan/client.py

def wait_for_result(
    self,
    uuid: str,
    timeout: float = 60.0,
    interval: float = 1.0,
    initial_wait: float | None = 10.0,
) -> None:
    """Wait for a scan result to be available.

    Args:
        uuid (str): UUID of a result.
        timeout (float, optional): Timeout in seconds. Defaults to 60.0.
        interval (float, optional): Interval in seconds. Defaults to 1.0.
        initial_wait (float | None, optional): Initial wait time in seconds. Set None to disable. Defaults to 10.0.
    """
    session = self._get_session()
    req = session.build_request("HEAD", f"/api/v1/result/{uuid}/")

    scanned_at = self._scan_uuid_timestamp_memo.get(uuid)
    if scanned_at and initial_wait:
        elapsed = time.time() - scanned_at
        if elapsed < initial_wait:
            time.sleep(initial_wait - elapsed)

    start_time = time.time()
    while True:
        res = self._send_request(session, req)
        if res.status_code == 200:
            self._scan_uuid_timestamp_memo.pop(uuid, None)
            return

        if time.time() - start_time > timeout:
            raise TimeoutError("Timeout waiting for scan result.")

        time.sleep(interval)