Vai al contenuto

Riferimento per hub_sdk/base/api_client.py

Nota

Questo file è disponibile all'indirizzo https://github.com/ultralytics/ hub-sdk/blob/main/ hub_sdk/base/api_client .py. Se riscontri un problema, contribuisci a risolverlo inviando una Pull Request 🛠️. Grazie 🙏!



hub_sdk.base.api_client.APIClientError

Basi: Exception

Classe di eccezione personalizzata per gli errori del client API.

Attributi:

Nome Tipo Descrizione
message str

Un messaggio di errore leggibile dall'uomo.

status_code int

Il codice di stato HTTP associato all'errore, se disponibile.

Codice sorgente in hub_sdk/base/api_client.py
class APIClientError(Exception):
    """
    Custom exception class for API client errors.

    Attributes:
        message (str): A human-readable error message.
        status_code (int): The HTTP status code associated with the error, if available.
    """

    def __init__(self, message: str, status_code: Optional[int] = None):
        """
        Initialize the APIClientError instance.

        Args:
            message (str): A human-readable error message.
            status_code (int, optional): The HTTP status code associated with the error, if available.
        """
        super().__init__(message)
        self.status_code = status_code
        self.message = message

    def __str__(self) -> str:
        return f"{self.__class__.__name__}: {self.args[0]}"

__init__(message, status_code=None)

Inizializza l'istanza APIClientError.

Parametri:

Nome Tipo Descrizione Predefinito
message str

Un messaggio di errore leggibile dall'uomo.

richiesto
status_code int

Il codice di stato HTTP associato all'errore, se disponibile.

None
Codice sorgente in hub_sdk/base/api_client.py
def __init__(self, message: str, status_code: Optional[int] = None):
    """
    Initialize the APIClientError instance.

    Args:
        message (str): A human-readable error message.
        status_code (int, optional): The HTTP status code associated with the error, if available.
    """
    super().__init__(message)
    self.status_code = status_code
    self.message = message



hub_sdk.base.api_client.APIClient

Rappresenta un client API per effettuare richieste a un URL di base specificato.

Attributi:

Nome Tipo Descrizione
base_url str

L'URL di base dell'API.

headers (dict, None)

Intestazioni da includere in ogni richiesta.

logger Logger

Un'istanza del logger per scopi di registrazione.

Codice sorgente in hub_sdk/base/api_client.py
class APIClient:
    """
    Represents an API client for making requests to a specified base URL.

    Attributes:
        base_url (str): The base URL for the API.
        headers (dict, None): Headers to be included in each request.
        logger (logging.Logger): An instance of the logger for logging purposes.
    """

    def __init__(self, base_url: str, headers: Optional[Dict] = None):
        """
        Initialize an instance of the APIClient class.

        Args:
            base_url (str): The base URL for the API.
            headers (dict, optional): Headers to be included in each request.
        """
        self.base_url = base_url
        self.headers = headers
        self.logger = logger

    def _make_request(
        self,
        method: str,
        endpoint: str,
        data: Optional[Dict] = None,
        json: Optional[Dict] = None,
        params: Optional[Dict] = None,
        files: Optional[Dict] = None,
        stream: bool = False,
    ) -> Optional[requests.Response]:
        """
        Make an HTTP request to the API.

        Args:
            method (str): The HTTP method to use for the request (e.g., "GET", "POST").
            endpoint (str): The endpoint to append to the base URL for the request.
            data (dict, optional): Data to be sent in the request's body.
            json (dict, optional): JSON data to be sent in the request's body.
            params (dict, optional): Query parameters for the request.
            files (dict, optional): Files to be sent as part of the form data.
            stream (bool, optional): Whether to stream the response content.

        Returns:
            (Optional[requests.Response]): The response object from the HTTP request, None if it fails and
        HUB_EXCEPTIONS off.

        Raises:
            (APIClientError): If an error occurs during the request, this exception is raised with an appropriate
        message based on the HTTP status code.
        """
        # Overwrite the base url if a http url is submitted
        url = endpoint if endpoint.startswith("http") else self.base_url + endpoint

        kwargs = {"params": params, "files": files, "headers": self.headers, "stream": stream}

        # Determine the request data based on 'data' or 'json_data'
        if json is not None:
            kwargs["json"] = json
        else:
            kwargs["data"] = data

        try:
            response = requests.request(method, url, **kwargs)

            response.raise_for_status()
            return response
        except requests.exceptions.RequestException as e:
            status_code = None
            # To handle Timeout and ConnectionError exceptions
            if hasattr(e, "response") and e.response:
                status_code = e.response.status_code

            error_msg = ErrorHandler(status_code).handle()
            self.logger.error(error_msg)

            if not HUB_EXCEPTIONS:
                raise APIClientError(error_msg, status_code=status_code) from e

    def get(self, endpoint: str, params=None) -> Optional[requests.Response]:
        """
        Make a GET request to the API.

        Args:
            endpoint (str): The endpoint to append to the base URL for the request.
            params (dict, optional): Query parameters for the request.

        Returns:
            (Optional[requests.Response]): The response object from the HTTP GET request, None if it fails.
        """
        return self._make_request("GET", endpoint, params=params)

    def post(
        self,
        endpoint: str,
        data: Optional[Dict] = None,
        json: Optional[Dict] = None,
        files: Optional[Dict] = None,
        stream=False,
    ) -> Optional[requests.Response]:
        """
        Make a POST request to the API.

        Args:
            endpoint (str): The endpoint to append to the base URL for the request.
            data (dict, optional): Data to be sent in the request's body.
            json (dict, optional): JSON data to be sent in the request's body.
            files (dict, optional): Files to be included in the request, if any.
            stream (bool, optional): If True, the response content will be streamed.

        Returns:
            (Optional[requests.Response]): The response object from the HTTP POST request.
        """
        return self._make_request("POST", endpoint, data=data, json=json, files=files, stream=stream)

    def put(
        self, endpoint: str, data: Optional[Dict] = None, json: Optional[Dict] = None
    ) -> Optional[requests.Response]:
        """
        Make a PUT request to the API.

        Args:
            endpoint (str): The endpoint to append to the base URL for the request.
            data (Optional[Dict], optional): Data to be sent in the request's body.
            json (Optional[Dict], optional): JSON data to be sent in the request's body

        Returns:
            (Optional[requests.Response]): The response object from the HTTP PUT request.
        """
        return self._make_request("PUT", endpoint, data=data, json=json)

    def delete(self, endpoint: str, params: Optional[Dict] = None) -> Optional[requests.Response]:
        """
        Make a DELETE request to the API.

        Args:
            endpoint (str): The endpoint to append to the base URL for the request.
            params (dict, optional): Parameters to include in the request.

        Returns:
            (Optional[requests.Response]): The response object from the HTTP DELETE request, or None if it fails.
        """
        return self._make_request("DELETE", endpoint, params=params)

    def patch(
        self, endpoint: str, data: Optional[Dict] = None, json: Optional[Dict] = None
    ) -> Optional[requests.Response]:
        """
        Make a PATCH request to the API.

        Args:
            endpoint (str): The endpoint to append to the base URL for the request.
            data (dict, optional): Data to be sent in the request's body.
            json (dict, optional): JSON data to be sent in the request's body.

        Returns:
            (Optional[requests.Response]): The response object from the HTTP PATCH request, or None if it fails.
        """
        return self._make_request("PATCH", endpoint, data=data, json=json)

__init__(base_url, headers=None)

Inizializza un'istanza della classe APIClient.

Parametri:

Nome Tipo Descrizione Predefinito
base_url str

L'URL di base dell'API.

richiesto
headers dict

Intestazioni da includere in ogni richiesta.

None
Codice sorgente in hub_sdk/base/api_client.py
def __init__(self, base_url: str, headers: Optional[Dict] = None):
    """
    Initialize an instance of the APIClient class.

    Args:
        base_url (str): The base URL for the API.
        headers (dict, optional): Headers to be included in each request.
    """
    self.base_url = base_url
    self.headers = headers
    self.logger = logger

delete(endpoint, params=None)

Effettua una richiesta DELETE all'API.

Parametri:

Nome Tipo Descrizione Predefinito
endpoint str

L'endpoint da aggiungere all'URL di base della richiesta.

richiesto
params dict

Parametri da includere nella richiesta.

None

Restituzione:

Tipo Descrizione
Optional[Response]

L'oggetto di risposta della richiesta HTTP DELETE o Nessuno se fallisce.

Codice sorgente in hub_sdk/base/api_client.py
def delete(self, endpoint: str, params: Optional[Dict] = None) -> Optional[requests.Response]:
    """
    Make a DELETE request to the API.

    Args:
        endpoint (str): The endpoint to append to the base URL for the request.
        params (dict, optional): Parameters to include in the request.

    Returns:
        (Optional[requests.Response]): The response object from the HTTP DELETE request, or None if it fails.
    """
    return self._make_request("DELETE", endpoint, params=params)

get(endpoint, params=None)

Effettua una richiesta GET all'API.

Parametri:

Nome Tipo Descrizione Predefinito
endpoint str

L'endpoint da aggiungere all'URL di base della richiesta.

richiesto
params dict

Parametri della query per la richiesta.

None

Restituzione:

Tipo Descrizione
Optional[Response]

L'oggetto di risposta della richiesta HTTP GET, Nessuno se fallisce.

Codice sorgente in hub_sdk/base/api_client.py
def get(self, endpoint: str, params=None) -> Optional[requests.Response]:
    """
    Make a GET request to the API.

    Args:
        endpoint (str): The endpoint to append to the base URL for the request.
        params (dict, optional): Query parameters for the request.

    Returns:
        (Optional[requests.Response]): The response object from the HTTP GET request, None if it fails.
    """
    return self._make_request("GET", endpoint, params=params)

patch(endpoint, data=None, json=None)

Effettua una richiesta di PATCH all'API.

Parametri:

Nome Tipo Descrizione Predefinito
endpoint str

L'endpoint da aggiungere all'URL di base della richiesta.

richiesto
data dict

Dati da inviare nel corpo della richiesta.

None
json dict

Dati JSON da inviare nel corpo della richiesta.

None

Restituzione:

Tipo Descrizione
Optional[Response]

L'oggetto di risposta della richiesta HTTP PATCH o Nessuno se fallisce.

Codice sorgente in hub_sdk/base/api_client.py
def patch(
    self, endpoint: str, data: Optional[Dict] = None, json: Optional[Dict] = None
) -> Optional[requests.Response]:
    """
    Make a PATCH request to the API.

    Args:
        endpoint (str): The endpoint to append to the base URL for the request.
        data (dict, optional): Data to be sent in the request's body.
        json (dict, optional): JSON data to be sent in the request's body.

    Returns:
        (Optional[requests.Response]): The response object from the HTTP PATCH request, or None if it fails.
    """
    return self._make_request("PATCH", endpoint, data=data, json=json)

post(endpoint, data=None, json=None, files=None, stream=False)

Effettua una richiesta POST all'API.

Parametri:

Nome Tipo Descrizione Predefinito
endpoint str

L'endpoint da aggiungere all'URL di base della richiesta.

richiesto
data dict

Dati da inviare nel corpo della richiesta.

None
json dict

Dati JSON da inviare nel corpo della richiesta.

None
files dict

Eventuali file da includere nella richiesta.

None
stream bool

Se Vero, il contenuto della risposta sarà trasmesso in streaming.

False

Restituzione:

Tipo Descrizione
Optional[Response]

L'oggetto di risposta della richiesta HTTP POST.

Codice sorgente in hub_sdk/base/api_client.py
def post(
    self,
    endpoint: str,
    data: Optional[Dict] = None,
    json: Optional[Dict] = None,
    files: Optional[Dict] = None,
    stream=False,
) -> Optional[requests.Response]:
    """
    Make a POST request to the API.

    Args:
        endpoint (str): The endpoint to append to the base URL for the request.
        data (dict, optional): Data to be sent in the request's body.
        json (dict, optional): JSON data to be sent in the request's body.
        files (dict, optional): Files to be included in the request, if any.
        stream (bool, optional): If True, the response content will be streamed.

    Returns:
        (Optional[requests.Response]): The response object from the HTTP POST request.
    """
    return self._make_request("POST", endpoint, data=data, json=json, files=files, stream=stream)

put(endpoint, data=None, json=None)

Effettua una richiesta PUT all'API.

Parametri:

Nome Tipo Descrizione Predefinito
endpoint str

L'endpoint da aggiungere all'URL di base della richiesta.

richiesto
data Optional[Dict]

Dati da inviare nel corpo della richiesta.

None
json Optional[Dict]

Dati JSON da inviare nel corpo della richiesta

None

Restituzione:

Tipo Descrizione
Optional[Response]

L'oggetto di risposta della richiesta HTTP PUT.

Codice sorgente in hub_sdk/base/api_client.py
def put(
    self, endpoint: str, data: Optional[Dict] = None, json: Optional[Dict] = None
) -> Optional[requests.Response]:
    """
    Make a PUT request to the API.

    Args:
        endpoint (str): The endpoint to append to the base URL for the request.
        data (Optional[Dict], optional): Data to be sent in the request's body.
        json (Optional[Dict], optional): JSON data to be sent in the request's body

    Returns:
        (Optional[requests.Response]): The response object from the HTTP PUT request.
    """
    return self._make_request("PUT", endpoint, data=data, json=json)