Zum Inhalt springen

Referenz fĂŒr hub_sdk/base/crud_client.py

Hinweis

Diese Datei ist verfĂŒgbar unter https://github.com/ultralytics/ hub-sdk/blob/main/ hub_sdk/base/crud_client .py. Wenn du ein Problem entdeckst, hilf bitte mit, es zu beheben, indem du einen Pull Request đŸ› ïž einreichst. Vielen Dank 🙏!



hub_sdk.base.crud_client.CRUDClient

Basen: APIClient

Stellt einen CRUD-Client (Create, Read, Update, Delete) fĂŒr die Interaktion mit einer bestimmten Ressource dar.

Attribute:

Name Typ Beschreibung
name str

Der Name, der mit den CRUD-VorgÀngen verbunden ist (z. B. "Benutzer").

logger Logger

Eine Instanz des Loggers fĂŒr Protokollierungszwecke.

Quellcode in 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}: %s", 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}({id}): %s", 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}({id}): %s", 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}({id}): %s", 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}: %s", e)

__init__(base_endpoint, name, headers)

Initialisiere eine CRUDClient-Instanz.

Parameter:

Name Typ Beschreibung Standard
base_endpoint str

Die Basis-Endpunkt-URL fĂŒr die API.

erforderlich
name str

Der Name, der mit den CRUD-VorgÀngen verbunden ist (z. B. "Benutzer").

erforderlich
headers dict

Kopfzeilen, die in API-Anfragen enthalten sein mĂŒssen.

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

Erstelle eine neue EntitÀt mithilfe der API.

Parameter:

Name Typ Beschreibung Standard
data dict

Die Daten, die als Teil der Erstellungsanfrage gesendet werden sollen.

erforderlich

Retouren:

Typ Beschreibung
Optional[Response]

Antwortobjekt von der Erstellungsanfrage oder Keine, wenn der Upload fehlschlÀgt.

Quellcode in 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}: %s", e)

delete(id, hard=False)

Lösche eine EntitĂ€t ĂŒber die API.

Parameter:

Name Typ Beschreibung Standard
id str

Der eindeutige Bezeichner der zu löschenden EntitÀt.

erforderlich
hard bool

Wenn True, fĂŒhre eine harte Löschung durch. Wenn False, fĂŒhre eine weiche Löschung durch.

False

Retouren:

Typ Beschreibung
Optional[Response]

Antwortobjekt der Löschanfrage oder Keine, wenn das Löschen fehlschlÀgt.

Quellcode in 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}({id}): %s", e)

list(page=0, limit=10)

Listen Sie EntitÀten auf, die die API verwenden.

Parameter:

Name Typ Beschreibung Standard
page int

Die Seitenzahl, die abgerufen werden soll.

0
limit int

Die maximale Anzahl von EntitÀten pro Seite.

10

Retouren:

Typ Beschreibung
Optional[Response]

Antwortobjekt der Listenanfrage oder Keine, wenn sie fehlschlÀgt.

Quellcode in 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}: %s", e)

read(id)

Abrufen von Details zu einer bestimmten EntitÀt.

Parameter:

Name Typ Beschreibung Standard
id str

Der eindeutige Bezeichner der abzurufenden EntitÀt.

erforderlich

Retouren:

Typ Beschreibung
Optional[Response]

Antwortobjekt der Leseanfrage oder Keine, wenn das Lesen fehlschlÀgt.

Quellcode in 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}({id}): %s", e)

update(id, data)

Aktualisiere eine bestehende EntitÀt mithilfe der API.

Parameter:

Name Typ Beschreibung Standard
id str

Der eindeutige Bezeichner der zu aktualisierenden EntitÀt.

erforderlich
data dict

Die aktualisierten Daten, die in der Aktualisierungsanfrage gesendet werden sollen.

erforderlich

Retouren:

Typ Beschreibung
Optional[Response]

Antwortobjekt der Aktualisierungsanfrage oder Keine, wenn die Aktualisierung fehlschlÀgt.

Quellcode in 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}({id}): %s", e)