Zum Inhalt springen

Referenz für hub_sdk/base/api_client.py

Hinweis

Diese Datei ist verfügbar unter https://github.com/ultralytics/hub-sdk/blob/main/hub_sdk/base/api_client .py. Wenn du ein Problem entdeckst, hilf bitte, es zu beheben, indem du einen Pull Request 🛠️ einreichst. Vielen Dank 🙏!



hub_sdk.base.api_client.APIClientError

Basen: Exception

Benutzerdefinierte Ausnahmeklasse für API-Client-Fehler.

Attribute:

Name Typ Beschreibung
message str

Eine für Menschen lesbare Fehlermeldung.

status_code int

Der mit dem Fehler verbundene HTTP-Statuscode, falls verfügbar.

Quellcode 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)

Initialisiere die APIClientError-Instanz.

Parameter:

Name Typ Beschreibung Standard
message str

Eine für Menschen lesbare Fehlermeldung.

erforderlich
status_code int

Der mit dem Fehler verbundene HTTP-Statuscode, falls verfügbar.

None
Quellcode 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

Stellt einen API-Client dar, der Anfragen an eine bestimmte Basis-URL stellt.

Attribute:

Name Typ Beschreibung
base_url str

Die Basis-URL für die API.

headers (dict, None)

Kopfzeilen, die in jeder Anfrage enthalten sein müssen.

logger Logger

Eine Instanz des Loggers für Protokollierungszwecke.

Quellcode 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 is not None:
                status_code = e.response.status_code

            error_msg = ErrorHandler(status_code, headers=response.headers).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)

Initialisiere eine Instanz der APIClient-Klasse.

Parameter:

Name Typ Beschreibung Standard
base_url str

Die Basis-URL für die API.

erforderlich
headers dict

Kopfzeilen, die in jeder Anfrage enthalten sein müssen.

None
Quellcode 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)

Stelle eine DELETE-Anfrage an die API.

Parameter:

Name Typ Beschreibung Standard
endpoint str

Der Endpunkt, der an die Basis-URL für die Anfrage angehängt wird.

erforderlich
params dict

Parameter, die in der Anfrage enthalten sein müssen.

None

Retouren:

Typ Beschreibung
Optional[Response]

Das Antwortobjekt der HTTP-DELETE-Anfrage oder Keine, wenn sie fehlschlägt.

Quellcode 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)

Stelle eine GET-Anfrage an die API.

Parameter:

Name Typ Beschreibung Standard
endpoint str

Der Endpunkt, der an die Basis-URL für die Anfrage angehängt wird.

erforderlich
params dict

Abfrageparameter für die Anfrage.

None

Retouren:

Typ Beschreibung
Optional[Response]

Das Antwortobjekt der HTTP-GET-Anfrage, keine, wenn sie fehlschlägt.

Quellcode 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)

Stelle eine PATCH-Anfrage an die API.

Parameter:

Name Typ Beschreibung Standard
endpoint str

Der Endpunkt, der an die Basis-URL für die Anfrage angehängt wird.

erforderlich
data dict

Daten, die im Body der Anfrage gesendet werden.

None
json dict

JSON-Daten, die im Body der Anfrage gesendet werden sollen.

None

Retouren:

Typ Beschreibung
Optional[Response]

Das Antwortobjekt der HTTP PATCH-Anfrage oder Keine, wenn sie fehlschlägt.

Quellcode 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)

Stelle eine POST-Anfrage an die API.

Parameter:

Name Typ Beschreibung Standard
endpoint str

Der Endpunkt, der an die Basis-URL für die Anfrage angehängt wird.

erforderlich
data dict

Daten, die im Body der Anfrage gesendet werden.

None
json dict

JSON-Daten, die im Body der Anfrage gesendet werden sollen.

None
files dict

Dateien, die dem Antrag beigefügt werden sollen, falls vorhanden.

None
stream bool

Wenn True, wird der Inhalt der Antwort gestreamt.

False

Retouren:

Typ Beschreibung
Optional[Response]

Das Antwortobjekt der HTTP-POST-Anfrage.

Quellcode 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)

Stelle eine PUT-Anfrage an die API.

Parameter:

Name Typ Beschreibung Standard
endpoint str

Der Endpunkt, der an die Basis-URL für die Anfrage angehängt wird.

erforderlich
data Optional[Dict]

Daten, die im Body der Anfrage gesendet werden.

None
json Optional[Dict]

JSON-Daten, die im Body der Anfrage gesendet werden

None

Retouren:

Typ Beschreibung
Optional[Response]

Das Antwortobjekt der HTTP PUT-Anfrage.

Quellcode 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)