Skip to content

Reference for hub_sdk/base/crud_client.py

Note

This file is available at https://github.com/ultralytics/hub-sdk/blob/main/hub_sdk/base/crud_client.py. If you spot a problem please help fix it by contributing a Pull Request 🛠️. Thank you 🙏!


hub_sdk.base.crud_client.CRUDClient 

CRUDClient(base_endpoint: str, name: str, headers: dict)

Bases: APIClient

Represents a CRUD (Create, Read, Update, Delete) client for interacting with a specific resource.

This class provides methods for performing standard CRUD operations on API resources, handling errors gracefully and providing logging for failed operations.

Attributes:

Name Type Description
name str

The name associated with the CRUD operations (e.g., "User").
logger Logger

An instance of the logger for logging purposes.

Parameters:

Name Type Description Default
base_endpoint str

The base endpoint URL for the API.

 required
name str

The name associated with the CRUD operations (e.g., "User").

 required
headers dict

Headers to be included in API requests.

 required
Source code in hub_sdk/base/crud_client.py 
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(self, base_endpoint: str, name: str, headers: dict):
    """
    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 

create(data: dict) -> Optional[Response]

Create a new entity using the API.

Parameters:

Name Type Description Default
data dict

The data to be sent as part of the creation request.

 required

Returns:

Type Description
Optional[Response]

Response object from the create request, or None if creation fails.
Source code in hub_sdk/base/crud_client.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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 creation fails.
    """
    try:
        return self.post("", json=data)
    except Exception as e:
        self.logger.error(f"Failed to create {self.name}: {e}")

delete 

delete(id: str, hard: bool = False) -> Optional[Response]

Delete an entity using the API.

Parameters:

Name Type Description Default
id str

The unique identifier of the entity to delete.

 required
hard bool

If True, perform a hard delete. If False, perform a soft delete.

 False

Returns:

Type Description
Optional[Response]

Response object from the delete request, or None if deletion fails.
Source code in hub_sdk/base/crud_client.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
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 deletion 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 

list(page: int = 0, limit: int = 10) -> Optional[Response]

List entities using the API with pagination support.

Parameters:

Name Type Description Default
page int

The page number to retrieve.

 0
limit int

The maximum number of entities per page.

 10

Returns:

Type Description
Optional[Response]

Response object from the list request, or None if listing fails.
Source code in hub_sdk/base/crud_client.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def list(self, page: int = 0, limit: int = 10) -> Optional[Response]:
    """
    List entities using the API with pagination support.

    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 listing 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 

read(id: str) -> Optional[Response]

Retrieve details of a specific entity.

Parameters:

Name Type Description Default
id str

The unique identifier of the entity to retrieve.

 required

Returns:

Type Description
Optional[Response]

Response object from the read request, or None if retrieval fails.
Source code in hub_sdk/base/crud_client.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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 retrieval 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 

update(id: str, data: dict) -> Optional[Response]

Update an existing entity using the API.

Parameters:

Name Type Description Default
id str

The unique identifier of the entity to update.

 required
data dict

The updated data to be sent in the update request.

 required

Returns:

Type Description
Optional[Response]

Response object from the update request, or None if update fails.
Source code in hub_sdk/base/crud_client.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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}")



📅 Created 1 year ago ✏️ Updated 1 month ago
glenn-jocher