Class Api

botasaurus_api/botasaurus_api/__init__.py:23–325

            response.raise_for_status()

class Api:
    def __init__(self, api_url: Optional[str] = None, create_response_files: bool = True) -> None:
        """
        Initializes the API client with a specified server URL and an option to create response files.
        
        :param api_url: The base URL for the API server. If not specified, defaults to "http://localhost:8000".
        :param create_response_files: Indicates if the client should create response files for each API call. This is useful for debugging or development purposes. Defaults to True.
        """
        DEFAULT_API_URL = "http://localhost:8000"
        self._api_url = remove_after_first_slash(api_url) if api_url else  DEFAULT_API_URL
        self._create_response_files = create_response_files  # Flag for creating response files
        if not self.is_api_running():
            raise ApiException(f"API at {self._api_url} is not running. Please check if the API is up and running.")

    def _write_json(self, filename, data):
        """
        Writes data to a JSON file specified by the filename. This method only runs if the create_response_files flag is True.

        :param filename: The filename for the JSON file to be created.
        :param data: The data to be written to the file.
        """
        if self._create_response_files:  
            path = _create_filename(filename)
            write_json_response(path, data) 
            print(f"View {filename} response at: ./{path}")

    def _make_api_url(self, path):
        return f"{self._api_url}/{path}"

        
    def is_api_running(self) -> bool:
        """
        Checks if the API is running by performing a health check on the "/api" endpoint.

        :return: True if the health check is successful, otherwise False.
        """
        try:
            response = requests.get(self._make_api_url("api"))
            # Check if the response status code is 200 (OK)
            return response.status_code == 200
        except ConnectionError:
            raise ApiException("""API at {} is not running. 
Check the network connection, or verify if the API is running on a different endpoint. In case the API is running on a different endpoint, you can pass the endpoint as follows:
api = Api('https://example.com')""".format(self._api_url))

    def create_async_task(self, data, scraper_name=None):
        """
        Submits an asynchronous task to the server.

        :param data: The data to be processed by the task.
        :param scraper_name: The name of the scraper to use for the task. If not provided, the server will use the default scraper.
        :return: The created task object.
        """
        url = self._make_api_url("api/tasks/create-task-async")
        payload = {
            "data": data,
            "scraper_name": scraper_name,
        }