Vai al contenuto

Riferimento per hub_sdk/base/crud_client.py

Nota

Questo file è disponibile all'indirizzo https://github.com/ultralytics/ hub-sdk/blob/main/ hub_sdk/base/crud_client .py. Se riscontri un problema, contribuisci a risolverlo inviando una Pull Request 🛠️. Grazie 🙏!



hub_sdk.base.crud_client.CRUDClient

Basi: APIClient

Rappresenta un client CRUD (Create, Read, Update, Delete) per interagire con una risorsa specifica.

Attributi:

Nome Tipo Descrizione
name str

Il nome associato alle operazioni CRUD (ad esempio, "Utente").

logger Logger

Un'istanza del logger per scopi di registrazione.

Codice sorgente 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)

Inizializza un'istanza di CRUDClient.

Parametri:

Nome Tipo Descrizione Predefinito
base_endpoint str

L'URL dell'endpoint di base dell'API.

richiesto
name str

Il nome associato alle operazioni CRUD (ad esempio, "Utente").

richiesto
headers dict

Intestazioni da includere nelle richieste API.

richiesto
Codice sorgente 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)

Crea una nuova entità utilizzando l'API.

Parametri:

Nome Tipo Descrizione Predefinito
data dict

I dati da inviare come parte della richiesta di creazione.

richiesto

Restituzione:

Tipo Descrizione
Optional[Response]

Oggetto di risposta della richiesta di creazione, oppure Nessuno se il caricamento fallisce.

Codice sorgente 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)

Elimina un'entità utilizzando l'API.

Parametri:

Nome Tipo Descrizione Predefinito
id str

L'identificativo univoco dell'entità da eliminare.

richiesto
hard bool

Se Vero, esegui una cancellazione definitiva. Se Falso, esegui una cancellazione morbida.

False

Restituzione:

Tipo Descrizione
Optional[Response]

Oggetto di risposta della richiesta di cancellazione, oppure Nessuno se la cancellazione fallisce.

Codice sorgente 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)

Elenca le entità che utilizzano l'API.

Parametri:

Nome Tipo Descrizione Predefinito
page int

Il numero di pagina da recuperare.

0
limit int

Il numero massimo di entità per pagina.

10

Restituzione:

Tipo Descrizione
Optional[Response]

Oggetto di risposta della richiesta di elenco, oppure None se fallisce.

Codice sorgente 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)

Recupera i dettagli di un'entità specifica.

Parametri:

Nome Tipo Descrizione Predefinito
id str

L'identificatore unico dell'entità da recuperare.

richiesto

Restituzione:

Tipo Descrizione
Optional[Response]

Oggetto di risposta della richiesta di lettura, oppure Nessuno se la lettura fallisce.

Codice sorgente 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)

Aggiorna un'entità esistente utilizzando l'API.

Parametri:

Nome Tipo Descrizione Predefinito
id str

L'identificatore unico dell'entità da aggiornare.

richiesto
data dict

I dati aggiornati da inviare nella richiesta di aggiornamento.

richiesto

Restituzione:

Tipo Descrizione
Optional[Response]

Oggetto di risposta della richiesta di aggiornamento, oppure Nessuno se l'aggiornamento fallisce.

Codice sorgente 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)