Bỏ để qua phần nội dung

Tài liệu tham khảo cho hub_sdk/base/crud_client.py

Ghi

Tệp này có sẵn tại https://github.com/ultralytics/hub-sdk/blob/main/hub_sdk/base/crud_client.py. Nếu bạn phát hiện ra một vấn đề, vui lòng giúp khắc phục nó bằng cách đóng góp Yêu cầu 🛠️ kéo. Cảm ơn bạn 🙏 !



hub_sdk.base.crud_client.CRUDClient

Căn cứ: APIClient

Đại diện cho một máy khách CRUD (Tạo, Đọc, Cập nhật, Xóa) để tương tác với một tài nguyên cụ thể.

Thuộc tính:

Tên Kiểu Sự miêu tả
name str

Tên được liên kết với các hoạt động CRUD (ví dụ: "Người dùng").

logger Logger

Một ví dụ của logger cho mục đích ghi nhật ký.

Mã nguồn trong hub_sdk/base/crud_client.py
class CRUDClient(APIClient):
    """
    Represents a CRUD (Create, Read, Update, Delete) client for interacting with a specific resource.

    Attributes:
        name (str): The name associated with the CRUD operations (e.g., "User").
        logger (logging.Logger): An instance of the logger for logging purposes.
    """

    def __init__(self, base_endpoint, name, headers):
        """
        Initialize a CRUDClient instance.

        Args:
            base_endpoint (str): The base endpoint URL for the API.
            name (str): The name associated with the CRUD operations (e.g., "User").
            headers (dict): Headers to be included in API requests.
        """
        super().__init__(f"{HUB_FUNCTIONS_ROOT}/v1/{base_endpoint}", headers)
        self.name = name
        self.logger = logger

    def create(self, data: dict) -> Optional[Response]:
        """
        Create a new entity using the API.

        Args:
            data (dict): The data to be sent as part of the creation request.

        Returns:
            (Optional[Response]): Response object from the create request, or None if upload fails.
        """
        try:
            return self.post("", json=data)
        except Exception as e:
            self.logger.error(f"Failed to create {self.name}: {e}")

    def read(self, id: str) -> Optional[Response]:
        """
        Retrieve details of a specific entity.

        Args:
            id (str): The unique identifier of the entity to retrieve.

        Returns:
            (Optional[Response]): Response object from the read request, or None if read fails.
        """
        try:
            return self.get(f"/{id}")
        except Exception as e:
            self.logger.error(f"Failed to read {self.name} with ID: {id}, {e}")

    def update(self, id: str, data: dict) -> Optional[Response]:
        """
        Update an existing entity using the API.

        Args:
            id (str): The unique identifier of the entity to update.
            data (dict): The updated data to be sent in the update request.

        Returns:
            (Optional[Response]): Response object from the update request, or None if update fails.
        """
        try:
            return self.patch(f"/{id}", json=data)
        except Exception as e:
            self.logger.error(f"Failed to update {self.name} with ID: {id}, {e}")

    def delete(self, id: str, hard: bool = False) -> Optional[Response]:
        """
        Delete an entity using the API.

        Args:
            id (str): The unique identifier of the entity to delete.
            hard (bool, optional): If True, perform a hard delete. If False, perform a soft delete.

        Returns:
            (Optional[Response]): Response object from the delete request, or None if delete fails.
        """
        try:
            return super().delete(f"/{id}", {"hard": hard})
        except Exception as e:
            self.logger.error(f"Failed to delete {self.name} with ID: {id}, {e}")

    def list(self, page: int = 0, limit: int = 10) -> Optional[Response]:
        """
        List entities using the API.

        Args:
            page (int, optional): The page number to retrieve.
            limit (int, optional): The maximum number of entities per page.

        Returns:
            (Optional[Response]): Response object from the list request, or None if it fails.
        """
        try:
            params = {"page": page, "limit": limit}
            return self.get("", params=params)
        except Exception as e:
            self.logger.error(f"Failed to list {self.name}: {e}")

__init__(base_endpoint, name, headers)

Khởi tạo phiên bản CRUDClient.

Thông số:

Tên Kiểu Sự miêu tả Mặc định
base_endpoint str

URL điểm cuối cơ sở cho API.

bắt buộc
name str

Tên được liên kết với các hoạt động CRUD (ví dụ: "Người dùng").

bắt buộc
headers dict

Tiêu đề được bao gồm trong yêu cầu API.

bắt buộc
Mã nguồn trong hub_sdk/base/crud_client.py
def __init__(self, base_endpoint, name, headers):
    """
    Initialize a CRUDClient instance.

    Args:
        base_endpoint (str): The base endpoint URL for the API.
        name (str): The name associated with the CRUD operations (e.g., "User").
        headers (dict): Headers to be included in API requests.
    """
    super().__init__(f"{HUB_FUNCTIONS_ROOT}/v1/{base_endpoint}", headers)
    self.name = name
    self.logger = logger

create(data)

Tạo một thực thể mới bằng API.

Thông số:

Tên Kiểu Sự miêu tả Mặc định
data dict

Dữ liệu sẽ được gửi như một phần của yêu cầu tạo.

bắt buộc

Trở lại:

Kiểu Sự miêu tả
Optional[Response]

Đối tượng phản hồi từ yêu cầu tạo hoặc Không có nếu tải lên không thành công.

Mã nguồn trong hub_sdk/base/crud_client.py
def create(self, data: dict) -> Optional[Response]:
    """
    Create a new entity using the API.

    Args:
        data (dict): The data to be sent as part of the creation request.

    Returns:
        (Optional[Response]): Response object from the create request, or None if upload fails.
    """
    try:
        return self.post("", json=data)
    except Exception as e:
        self.logger.error(f"Failed to create {self.name}: {e}")

delete(id, hard=False)

Xóa một thực thể bằng API.

Thông số:

Tên Kiểu Sự miêu tả Mặc định
id str

Mã định danh duy nhất của thực thể cần xóa.

bắt buộc
hard bool

Nếu True, hãy thực hiện xóa cứng. Nếu False, hãy thực hiện xóa mềm.

False

Trở lại:

Kiểu Sự miêu tả
Optional[Response]

Phản hồi đối tượng từ yêu cầu xóa, hoặc Không có nếu xóa không thành công.

Mã nguồn trong hub_sdk/base/crud_client.py
def delete(self, id: str, hard: bool = False) -> Optional[Response]:
    """
    Delete an entity using the API.

    Args:
        id (str): The unique identifier of the entity to delete.
        hard (bool, optional): If True, perform a hard delete. If False, perform a soft delete.

    Returns:
        (Optional[Response]): Response object from the delete request, or None if delete fails.
    """
    try:
        return super().delete(f"/{id}", {"hard": hard})
    except Exception as e:
        self.logger.error(f"Failed to delete {self.name} with ID: {id}, {e}")

list(page=0, limit=10)

Liệt kê các thực thể bằng API.

Thông số:

Tên Kiểu Sự miêu tả Mặc định
page int

Số trang cần truy xuất.

0
limit int

Số lượng thực thể tối đa trên mỗi trang.

10

Trở lại:

Kiểu Sự miêu tả
Optional[Response]

Đối tượng phản hồi từ yêu cầu danh sách hoặc Không có nếu không thành công.

Mã nguồn trong hub_sdk/base/crud_client.py
def list(self, page: int = 0, limit: int = 10) -> Optional[Response]:
    """
    List entities using the API.

    Args:
        page (int, optional): The page number to retrieve.
        limit (int, optional): The maximum number of entities per page.

    Returns:
        (Optional[Response]): Response object from the list request, or None if it fails.
    """
    try:
        params = {"page": page, "limit": limit}
        return self.get("", params=params)
    except Exception as e:
        self.logger.error(f"Failed to list {self.name}: {e}")

read(id)

Truy xuất thông tin chi tiết của một thực thể cụ thể.

Thông số:

Tên Kiểu Sự miêu tả Mặc định
id str

Mã định danh duy nhất của thực thể cần truy xuất.

bắt buộc

Trở lại:

Kiểu Sự miêu tả
Optional[Response]

Đối tượng phản hồi từ yêu cầu đọc hoặc Không có nếu đọc không thành công.

Mã nguồn trong hub_sdk/base/crud_client.py
def read(self, id: str) -> Optional[Response]:
    """
    Retrieve details of a specific entity.

    Args:
        id (str): The unique identifier of the entity to retrieve.

    Returns:
        (Optional[Response]): Response object from the read request, or None if read fails.
    """
    try:
        return self.get(f"/{id}")
    except Exception as e:
        self.logger.error(f"Failed to read {self.name} with ID: {id}, {e}")

update(id, data)

Cập nhật thực thể hiện có bằng API.

Thông số:

Tên Kiểu Sự miêu tả Mặc định
id str

Mã định danh duy nhất của thực thể cần cập nhật.

bắt buộc
data dict

Dữ liệu cập nhật sẽ được gửi trong yêu cầu cập nhật.

bắt buộc

Trở lại:

Kiểu Sự miêu tả
Optional[Response]

Phản hồi đối tượng từ yêu cầu cập nhật hoặc Không có nếu cập nhật không thành công.

Mã nguồn trong hub_sdk/base/crud_client.py
def update(self, id: str, data: dict) -> Optional[Response]:
    """
    Update an existing entity using the API.

    Args:
        id (str): The unique identifier of the entity to update.
        data (dict): The updated data to be sent in the update request.

    Returns:
        (Optional[Response]): Response object from the update request, or None if update fails.
    """
    try:
        return self.patch(f"/{id}", json=data)
    except Exception as e:
        self.logger.error(f"Failed to update {self.name} with ID: {id}, {e}")