跳至内容

参考资料 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)