コンテンツへスキップ

参考 hub_sdk/base/api_client.py

備考

このファイルはhttps://github.com/ultralytics/hub-sdk/blob/main/ hub_sdk/base/api_client .py にあります。もし問題を発見したら、Pull Request🛠️ を投稿して修正にご協力ください。ありがとうございました!



hub_sdk.base.api_client.APIClientError

ベース: Exception

API クライアントエラー用のカスタム例外クラス。

属性:

名称 タイプ 説明
message str

人間が読めるエラーメッセージ。

status_code int

もしあれば、エラーに関連するHTTPステータスコード。

ソースコード 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)

APIClientErrorインスタンスを初期化します。

パラメーター

名称 タイプ 説明 デフォルト
message str

人間が読めるエラーメッセージ。

必須
status_code int

もしあれば、エラーに関連するHTTPステータスコード。

None
ソースコード 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

指定されたベースURLへのリクエストを行うAPIクライアントを表す。

属性:

名称 タイプ 説明
base_url str

APIのベースURL。

headers (dict, None)

各リクエストに含まれるヘッダー。

logger Logger

ロギング用のロガーのインスタンス。

ソースコード 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)

APIClientクラスのインスタンスを初期化します。

パラメーター

名称 タイプ 説明 デフォルト
base_url str

APIのベースURL。

必須
headers dict

各リクエストに含まれるヘッダー。

None
ソースコード 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)

APIにDELETEリクエストを行う。

パラメーター

名称 タイプ 説明 デフォルト
endpoint str

リクエストのベース URL に追加するエンドポイント。

必須
params dict

リクエストに含めるパラメータ。

None

リターンズ

タイプ 説明
Optional[Response]

HTTP DELETE リクエストのレスポンス・オブジェクト。

ソースコード 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)

APIにGETリクエストを行う。

パラメーター

名称 タイプ 説明 デフォルト
endpoint str

リクエストのベース URL に追加するエンドポイント。

必須
params dict

リクエストのクエリパラメータ。

None

リターンズ

タイプ 説明
Optional[Response]

HTTP GETリクエストのレスポンス・オブジェクト。

ソースコード 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)

APIにPATCHリクエストを行う。

パラメーター

名称 タイプ 説明 デフォルト
endpoint str

リクエストのベース URL に追加するエンドポイント。

必須
data dict

リクエストのボディに送られるデータ。

None
json dict

リクエスト・ボディに送信するJSONデータ。

None

リターンズ

タイプ 説明
Optional[Response]

HTTP PATCH リクエストのレスポンス・オブジェクト。

ソースコード 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)

APIにPOSTリクエストを行う。

パラメーター

名称 タイプ 説明 デフォルト
endpoint str

リクエストのベース URL に追加するエンドポイント。

必須
data dict

リクエストのボディに送られるデータ。

None
json dict

リクエスト・ボディに送信するJSONデータ。

None
files dict

リクエストに含まれるファイル(もしあれば)。

None
stream bool

Trueの場合、レスポンスの内容はストリームされます。

False

リターンズ

タイプ 説明
Optional[Response]

HTTP POSTリクエストのレスポンス・オブジェクト。

ソースコード 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)

APIにPUTリクエストを行う。

パラメーター

名称 タイプ 説明 デフォルト
endpoint str

リクエストのベース URL に追加するエンドポイント。

必須
data Optional[Dict]

リクエストのボディに送られるデータ。

None
json Optional[Dict]

リクエスト・ボディに送信されるJSONデータ

None

リターンズ

タイプ 説明
Optional[Response]

HTTP PUT リクエストのレスポンス・オブジェクト。

ソースコード 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)