انتقل إلى المحتوى

مرجع ل hub_sdk/base/crud_client.py

ملاحظه

هذا الملف متاح في https://github.com/ultralytics/hub-SDK / نقطة / الرئيسية /hub_sdk / قاعدة / crud_client.py. إذا اكتشفت مشكلة ، فيرجى المساعدة في إصلاحها من خلال المساهمة في طلب 🛠️ سحب. شكرا لك 🙏!



hub_sdk.base.crud_client.CRUDClient

قواعد: APIClient

يمثل عميل CRUD (إنشاء ، قراءة ، تحديث ، حذف) للتفاعل مع مورد معين.

سمات:

اسم نوع وصف
name str

الاسم المرتبط بعمليات CRUD (على سبيل المثال ، "المستخدم").

logger Logger

مثيل للمسجل لأغراض التسجيل.

شفرة المصدر في 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)

تهيئة مثيل CRUDClient.

البارامترات:

اسم نوع وصف افتراضي
base_endpoint str

عنوان URL لنقطة النهاية الأساسية لواجهة برمجة التطبيقات.

مطلوب
name str

الاسم المرتبط بعمليات CRUD (على سبيل المثال ، "المستخدم").

مطلوب
headers dict

الرؤوس التي سيتم تضمينها في طلبات واجهة برمجة التطبيقات.

مطلوب
شفرة المصدر في hub_sdk/base/crud_client.py
19 20 21 22 23 24 25 2627282930
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)

إنشاء كيان جديد باستخدام واجهة برمجة التطبيقات.

البارامترات:

اسم نوع وصف افتراضي
data dict

البيانات المراد إرسالها كجزء من طلب الإنشاء.

مطلوب

ارجاع:

نوع وصف
Optional[Response]

كائن الاستجابة من طلب الإنشاء، أو لا شيء إذا فشل التحميل.

شفرة المصدر في 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)

حذف كيان باستخدام واجهة برمجة التطبيقات.

البارامترات:

اسم نوع وصف افتراضي
id str

المعرف الفريد للكيان المراد حذفه.

مطلوب
hard bool

إذا كان هذا صحيحا ، فقم بإجراء حذف ثابت. إذا كان خطأ، فقم بإجراء حذف بسيط.

False

ارجاع:

نوع وصف
Optional[Response]

كائن الاستجابة من طلب الحذف، أو لا شيء إذا فشل الحذف.

شفرة المصدر في hub_sdk/base/crud_client.py
78 79 80 81 82 83 84 85868788899091 92
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)

سرد الكيانات باستخدام واجهة برمجة التطبيقات.

البارامترات:

اسم نوع وصف افتراضي
page int

رقم الصفحة المراد استردادها.

0
limit int

الحد الأقصى لعدد الكيانات في كل صفحة.

10

ارجاع:

نوع وصف
Optional[Response]

كائن الاستجابة من طلب القائمة، أو لا شيء إذا فشل.

شفرة المصدر في hub_sdk/base/crud_client.py
94 95 96 97 98  99 100 101 102 103 104 105  106 107 108 109 
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)

استرجاع تفاصيل كيان معين.

البارامترات:

اسم نوع وصف افتراضي
id str

المعرف الفريد للكيان المراد استرداده.

مطلوب

ارجاع:

نوع وصف
Optional[Response]

كائن الاستجابة من طلب القراءة، أو لا شيء إذا فشلت القراءة.

شفرة المصدر في hub_sdk/base/crud_client.py
47 48 49 50 51 52 53 54 5556 57585960
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)

تحديث كيان موجود باستخدام واجهة برمجة التطبيقات.

البارامترات:

اسم نوع وصف افتراضي
id str

المعرف الفريد للكيان المراد تحديثه.

مطلوب
data dict

البيانات المحدثة التي سيتم إرسالها في طلب التحديث.

مطلوب

ارجاع:

نوع وصف
Optional[Response]

كائن الاستجابة من طلب التحديث، أو لا شيء إذا فشل التحديث.

شفرة المصدر في 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)