跳到内容

rg.Argilla

要从 Python 与 Argilla 服务器交互,您可以使用 Argilla 类。 Argilla 客户端用于创建、获取、更新和删除所有 Argilla 资源,例如工作区、用户、数据集和记录。

使用示例

在 Hugging Face Spaces 上部署 Argilla 服务器

要在 Hugging Face Spaces 上部署 Argilla,请使用 deploy_on_spaces 方法。

import argilla as rg

client = rg.Argilla.deploy_on_spaces(api_key="12345678")

连接到 Argilla 服务器

要连接到 Argilla 服务器,请实例化 Argilla 类并传递服务器的 api_urlapi_key 以进行身份验证。

import argilla as rg

client = rg.Argilla(
    api_url="https://argilla.example.com",
    api_key="my_api_key",
)

访问数据集、工作区和用户对象

Argilla 客户端提供对 Argilla 服务器的 DatasetWorkspaceUser 对象的访问。

my_dataset = client.datasets("my_dataset")

my_workspace = client.workspaces("my_workspace")

my_user = client.users("my_user")

然后可以与这些资源交互以访问其属性和方法。 例如,列出工作区中的所有数据集

for dataset in my_workspace.datasets:
    print(dataset.name)

Argilla

基类:APIClient, SpacesDeploymentMixin, NotebookHTMLReprMixin

Argilla API 客户端。 这是与 API 交互的主要入口点。

属性

名称 类型 描述
工作区 工作区

工作区的集合。

数据集 数据集

数据集的集合。

用户 用户

用户的集合。

用户

当前用户。

源代码位于 src/argilla/client.py
class Argilla(_api.APIClient, SpacesDeploymentMixin, NotebookHTMLReprMixin):
    """Argilla API client. This is the main entry point to interact with the API.

    Attributes:
        workspaces: A collection of workspaces.
        datasets: A collection of datasets.
        users: A collection of users.
        me: The current user.
    """

    # Default instance of Argilla
    _default_client: Optional["Argilla"] = None

    def __init__(
        self,
        api_url: Optional[str] = DEFAULT_HTTP_CONFIG.api_url,
        api_key: Optional[str] = DEFAULT_HTTP_CONFIG.api_key,
        timeout: int = DEFAULT_HTTP_CONFIG.timeout,
        retries: int = DEFAULT_HTTP_CONFIG.retries,
        **http_client_args,
    ) -> None:
        """Inits the `Argilla` client.

        Args:
            api_url: the URL of the Argilla API. If not provided, then the value will try
                to be set from `ARGILLA_API_URL` environment variable. Defaults to
                `"http://localhost:6900"`.
            api_key: the key to be used to authenticate in the Argilla API. If not provided,
                then the value will try to be set from `ARGILLA_API_KEY` environment variable.
                Defaults to `None`.
            timeout: the maximum time in seconds to wait for a request to the Argilla API
                to be completed before raising an exception. Defaults to `60`.
            retries: the number of times to retry the HTTP connection to the Argilla API
                before raising an exception. Defaults to `5`.
        """
        super().__init__(api_url=api_url, api_key=api_key, timeout=timeout, retries=retries, **http_client_args)

        self._set_default(self)

    @property
    def workspaces(self) -> "Workspaces":
        """A collection of workspaces on the server."""
        return Workspaces(client=self)

    @property
    def datasets(self) -> "Datasets":
        """A collection of datasets on the server."""
        return Datasets(client=self)

    @property
    def users(self) -> "Users":
        """A collection of users on the server."""
        return Users(client=self)

    @property
    def webhooks(self) -> "Webhooks":
        """A collection of webhooks on the server."""
        return Webhooks(client=self)

    @cached_property
    def me(self) -> "User":
        from argilla.users import User

        return User(client=self, _model=self.api.users.get_me())

    ############################
    # Private methods
    ############################

    @classmethod
    def _set_default(cls, client: "Argilla") -> None:
        """Set the default instance of Argilla."""
        cls._default_client = client

    @classmethod
    def _get_default(cls) -> "Argilla":
        """Get the default instance of Argilla. If it doesn't exist, create a new one."""
        if cls._default_client is None:
            cls._default_client = Argilla()
        return cls._default_client

workspaces: Workspaces property

服务器上工作区的集合。

datasets: Datasets property

服务器上数据集的集合。

users: Users property

服务器上用户的集合。

webhooks: Webhooks property

服务器上 Webhooks 的集合。

__init__(api_url=DEFAULT_HTTP_CONFIG.api_url, api_key=DEFAULT_HTTP_CONFIG.api_key, timeout=DEFAULT_HTTP_CONFIG.timeout, retries=DEFAULT_HTTP_CONFIG.retries, **http_client_args)

初始化 Argilla 客户端。

参数

名称 类型 描述 默认值
api_url 可选[str]

Argilla API 的 URL。 如果未提供,则该值将尝试从 ARGILLA_API_URL 环境变量中设置。 默认为 "http://localhost:6900"

api_url
api_key 可选[str]

用于在 Argilla API 中进行身份验证的密钥。 如果未提供,则该值将尝试从 ARGILLA_API_KEY 环境变量中设置。 默认为 None

api_key
timeout int

等待 Argilla API 请求完成的最长时间(秒),超过此时间将引发异常。 默认为 60

timeout
retries int

重试与 Argilla API 的 HTTP 连接的次数,超过此次数将引发异常。 默认为 5

retries
源代码位于 src/argilla/client.py
def __init__(
    self,
    api_url: Optional[str] = DEFAULT_HTTP_CONFIG.api_url,
    api_key: Optional[str] = DEFAULT_HTTP_CONFIG.api_key,
    timeout: int = DEFAULT_HTTP_CONFIG.timeout,
    retries: int = DEFAULT_HTTP_CONFIG.retries,
    **http_client_args,
) -> None:
    """Inits the `Argilla` client.

    Args:
        api_url: the URL of the Argilla API. If not provided, then the value will try
            to be set from `ARGILLA_API_URL` environment variable. Defaults to
            `"http://localhost:6900"`.
        api_key: the key to be used to authenticate in the Argilla API. If not provided,
            then the value will try to be set from `ARGILLA_API_KEY` environment variable.
            Defaults to `None`.
        timeout: the maximum time in seconds to wait for a request to the Argilla API
            to be completed before raising an exception. Defaults to `60`.
        retries: the number of times to retry the HTTP connection to the Argilla API
            before raising an exception. Defaults to `5`.
    """
    super().__init__(api_url=api_url, api_key=api_key, timeout=timeout, retries=retries, **http_client_args)

    self._set_default(self)

Users

基类:Sequence['User'], ResourceHTMLReprMixin

用户的集合。 它可用于创建新用户或获取现有用户。

源代码位于 src/argilla/client.py
class Users(Sequence["User"], ResourceHTMLReprMixin):
    """A collection of users. It can be used to create a new user or to get an existing one."""

    class _Iterator(GenericIterator["User"]):
        pass

    def __init__(self, client: "Argilla") -> None:
        self._client = client
        self._api = client.api.users

    @overload
    def __call__(self, username: str) -> Optional["User"]:
        """Get a user by username if exists. Otherwise, returns `None`"""
        ...

    @overload
    def __call__(self, id: Union[UUID, str]) -> Optional["User"]:
        """Get a user by id if exists. Otherwise, returns `None`"""
        ...

    def __call__(self, username: str = None, id: Union[str, UUID] = None) -> Optional["User"]:
        if not (username or id):
            raise ArgillaError("One of 'username' or 'id' must be provided")
        if username and id:
            warnings.warn("Only one of 'username' or 'id' must be provided. Using 'id'")
            username = None

        if id is not None:
            model = _get_model_by_id(self._api, id)
            if model:
                return self._from_model(model)  # noqa
            warnings.warn(f"User with id {id!r} not found.")
        else:
            for model in self._api.list():
                if model.username == username:
                    return self._from_model(model)
            warnings.warn(f"User with username {username!r} not found.")

    def __iter__(self):
        return self._Iterator(self.list())

    @overload
    @abstractmethod
    def __getitem__(self, index: int) -> "User": ...

    @overload
    @abstractmethod
    def __getitem__(self, index: slice) -> Sequence["User"]: ...

    def __getitem__(self, index):
        model = self._api.list()[index]
        return self._from_model(model)

    def __len__(self) -> int:
        return len(self._api.list())

    def add(self, user: "User") -> "User":
        """Add a new user to Argilla.

        Args:
            user: User object.

        Returns:
            User: The created user.
        """
        user._client = self._client
        return user.create()

    @overload
    def list(self) -> List["User"]: ...

    @overload
    def list(self, workspace: "Workspace") -> List["User"]: ...

    def list(self, workspace: Optional["Workspace"] = None) -> List["User"]:
        """List all users."""
        if workspace is not None:
            models = self._api.list_by_workspace_id(workspace.id)
        else:
            models = self._api.list()

        return [self._from_model(model) for model in models]

    ############################
    # Private methods
    ############################

    def _repr_html_(self) -> str:
        return self._represent_as_html(resources=self.list())

    def _from_model(self, model: UserModel) -> "User":
        from argilla.users import User

        return User(client=self._client, _model=model)

add(user)

向 Argilla 添加新用户。

参数

名称 类型 描述 默认值
user 用户

用户对象。

必需

返回值

名称 类型 描述
用户 用户

创建的用户。

源代码位于 src/argilla/client.py
def add(self, user: "User") -> "User":
    """Add a new user to Argilla.

    Args:
        user: User object.

    Returns:
        User: The created user.
    """
    user._client = self._client
    return user.create()

list(workspace=None)

列出所有用户。

源代码位于 src/argilla/client.py
def list(self, workspace: Optional["Workspace"] = None) -> List["User"]:
    """List all users."""
    if workspace is not None:
        models = self._api.list_by_workspace_id(workspace.id)
    else:
        models = self._api.list()

    return [self._from_model(model) for model in models]

Workspaces

基类:Sequence['Workspace'], ResourceHTMLReprMixin

工作区的集合。 它可用于创建新工作区或获取现有工作区。

源代码位于 src/argilla/client.py
class Workspaces(Sequence["Workspace"], ResourceHTMLReprMixin):
    """A collection of workspaces. It can be used to create a new workspace or to get an existing one."""

    class _Iterator(GenericIterator["Workspace"]):
        pass

    def __init__(self, client: "Argilla") -> None:
        self._client = client
        self._api = client.api.workspaces

    @overload
    def __call__(self, name: str) -> Optional["Workspace"]:
        """Get a workspace by name if exists. Otherwise, returns `None`"""
        ...

    @overload
    def __call__(self, id: Union[UUID, str]) -> Optional["Workspace"]:
        """Get a workspace by id if exists. Otherwise, returns `None`"""
        ...

    def __call__(self, name: str = None, id: Union[UUID, str] = None) -> Optional["Workspace"]:
        if not (name or id):
            raise ArgillaError("One of 'name' or 'id' must be provided")

        if name and id:
            warnings.warn("Only one of 'name' or 'id' must be provided. Using 'id'")
            name = None

        if id is not None:
            model = _get_model_by_id(self._api, id)
            if model:
                return self._from_model(model)  # noqa
            warnings.warn(f"Workspace with id {id!r} not found")
        else:
            for model in self._api.list():
                if model.name == name:
                    return self._from_model(model)  # noqa
            warnings.warn(f"Workspace with name {name!r} not found.")

    def __iter__(self):
        return self._Iterator(self.list())

    @overload
    @abstractmethod
    def __getitem__(self, index: int) -> "Workspace": ...

    @overload
    @abstractmethod
    def __getitem__(self, index: slice) -> Sequence["Workspace"]: ...

    def __getitem__(self, index) -> "Workspace":
        model = self._api.list()[index]
        return self._from_model(model)

    def __len__(self) -> int:
        return len(self._api.list())

    def add(self, workspace: "Workspace") -> "Workspace":
        """Add a new workspace to the Argilla platform.
        Args:
            workspace: Workspace object.

        Returns:
            Workspace: The created workspace.
        """
        workspace._client = self._client
        return workspace.create()

    def list(self) -> List["Workspace"]:
        return [self._from_model(model) for model in self._api.list()]

    ############################
    # Properties
    ############################

    @property
    def default(self) -> "Workspace":
        """The default workspace."""
        if len(self) == 0:
            raise ArgillaError("There are no workspaces created. Please create a new workspace first")
        return self[0]

    ############################
    # Private methods
    ############################

    def _repr_html_(self) -> str:
        return self._represent_as_html(resources=self.list())

    def _from_model(self, model: WorkspaceModel) -> "Workspace":
        from argilla.workspaces import Workspace

        return Workspace.from_model(client=self._client, model=model)

default: Workspace property

默认工作区。

add(workspace)

向 Argilla 平台添加新工作区。 参数:workspace:工作区对象。

返回值

名称 类型 描述
工作区 工作区

创建的工作区。

源代码位于 src/argilla/client.py
def add(self, workspace: "Workspace") -> "Workspace":
    """Add a new workspace to the Argilla platform.
    Args:
        workspace: Workspace object.

    Returns:
        Workspace: The created workspace.
    """
    workspace._client = self._client
    return workspace.create()

Datasets

基类:Sequence['Dataset'], ResourceHTMLReprMixin

数据集的集合。 它可用于创建新数据集或获取现有数据集。

源代码位于 src/argilla/client.py
class Datasets(Sequence["Dataset"], ResourceHTMLReprMixin):
    """A collection of datasets. It can be used to create a new dataset or to get an existing one."""

    class _Iterator(GenericIterator["Dataset"]):
        def __next__(self):
            dataset = super().__next__()
            return dataset.get()

    def __init__(self, client: "Argilla") -> None:
        self._client = client
        self._api = client.api.datasets

    @overload
    def __call__(self, name: str, workspace: Optional[Union["Workspace", str]] = None) -> Optional["Dataset"]:
        """Get a dataset by name and workspace if exists. Otherwise, returns `None`"""
        ...

    @overload
    def __call__(self, id: Union[UUID, str]) -> Optional["Dataset"]:
        """Get a dataset by id if exists. Otherwise, returns `None`"""
        ...

    def __call__(
        self, name: str = None, workspace: Optional[Union["Workspace", str]] = None, id: Union[UUID, str] = None
    ) -> Optional["Dataset"]:
        if not (name or id):
            raise ArgillaError("One of 'name' or 'id' must be provided")

        if name and id:
            warnings.warn("Only one of 'name' or 'id' must be provided. Using 'id'")
            name = None

        if id is not None:
            model = _get_model_by_id(self._api, id)
            if model:
                return self._from_model(model)  # noqa
            warnings.warn(f"Dataset with id {id!r} not found")
        else:
            workspace = workspace or self._client.workspaces.default
            if isinstance(workspace, str):
                workspace = self._client.workspaces(workspace)

            for dataset in workspace.datasets:
                if dataset.name == name:
                    return dataset.get()
            warnings.warn(f"Dataset with name {name!r} not found in workspace {workspace.name!r}")

    def __iter__(self):
        return self._Iterator([self._from_model(model) for model in self._api.list()])

    @overload
    @abstractmethod
    def __getitem__(self, index: int) -> "Dataset": ...

    @overload
    @abstractmethod
    def __getitem__(self, index: slice) -> Sequence["Dataset"]: ...

    def __getitem__(self, index) -> "Dataset":
        model = self._api.list()[index]
        return self._from_model(model).get()

    def __len__(self) -> int:
        return len(self._api.list())

    def add(self, dataset: "Dataset") -> "Dataset":
        """
        Add a new dataset to the Argilla platform

        Args:
            dataset: Dataset object.

        Returns:
            Dataset: The created dataset.
        """
        dataset._client = self._client
        dataset.create()

        return dataset

    def list(self) -> List["Dataset"]:
        return list(self)

    ############################
    # Private methods
    ############################

    def _repr_html_(self) -> str:
        return self._represent_as_html(resources=self.list())

    def _from_model(self, model: DatasetModel) -> "Dataset":
        from argilla.datasets import Dataset

        return Dataset.from_model(model=model, client=self._client)

add(dataset)

向 Argilla 平台添加新数据集

参数

名称 类型 描述 默认值
dataset 数据集

数据集对象。

必需

返回值

名称 类型 描述
数据集 数据集

创建的数据集。

源代码位于 src/argilla/client.py
def add(self, dataset: "Dataset") -> "Dataset":
    """
    Add a new dataset to the Argilla platform

    Args:
        dataset: Dataset object.

    Returns:
        Dataset: The created dataset.
    """
    dataset._client = self._client
    dataset.create()

    return dataset

SpacesDeploymentMixin

基类:LoggingMixin

源代码位于 src/argilla/_helpers/_deploy.py
class SpacesDeploymentMixin(LoggingMixin):
    @classmethod
    def deploy_on_spaces(
        cls,
        api_key: str,
        repo_name: Optional[str] = "argilla",
        org_name: Optional[str] = None,
        hf_token: Optional[str] = None,
        space_storage: Optional[Union[str, "SpaceStorage", Literal["small", "medium", "large"]]] = None,
        space_hardware: Optional[Union[str, "SpaceHardware", Literal["cpu-basic", "cpu-upgrade"]]] = "cpu-basic",
        private: Optional[Union[bool, None]] = False,
    ) -> "Argilla":
        """
        Deploys Argilla on Hugging Face Spaces.

        Args:
            api_key (str): The Argilla API key to be defined for the owner user and creator of the Space.
            repo_name (Optional[str]): The ID of the repository where Argilla will be deployed. Defaults to "argilla".
            org_name (Optional[str]): The name of the organization where Argilla will be deployed. Defaults to None.
            hf_token (Optional[Union[str, None]]): The Hugging Face authentication token. Defaults to None.
            space_storage (Optional[Union[str, SpaceStorage]]): The persistent storage size for the space. Defaults to None without persistent storage.
            space_hardware (Optional[Union[str, SpaceHardware]]): The hardware configuration for the space. Defaults to "cpu-basic" with downtime after 48 hours of inactivity.
            private (Optional[Union[bool, None]]): Whether the space should be private. Defaults to False.

        Returns:
            Argilla: The Argilla client.

        Example:
            ```Python
            import argilla as rg
            client = rg.Argilla.deploy_on_spaces(api_key="12345678")
            ```
        """
        hf_token = cls._acquire_hf_token(ht_token=hf_token)
        hf_api = HfApi(token=hf_token)

        # Get the org name from the repo name or default to the current user
        token_username = hf_api.whoami()["name"]
        org_name = org_name or token_username
        repo_id = f"{org_name}/{repo_name}"

        # Define the api_key for the space
        secrets = [
            {"key": "API_KEY", "value": api_key, "description": "The API key of the owner user."},
            {"key": "WORKSPACE", "value": "argilla", "description": "The workspace of the space."},
        ]

        # check API key length
        if len(api_key) < 8:
            raise ValueError(
                "Provided API key has invalid length. Please provide an apikey with at least 8 characters."
            )

        # Check if the space already exists
        if hf_api.repo_exists(repo_id=repo_id, repo_type="space"):
            if cls._is_space_stopped(hf_api.get_space_runtime(repo_id=repo_id).stage):
                hf_api.restart_space(repo_id=repo_id)
            warnings.warn(
                f"Space {repo_id} already exists. Using provided API key. If client authentication fails, go to "
                f"https://hugging-face.cn/spaces/{repo_id} to login with OAuth and get the correct API key.",
                stacklevel=2,
            )
        else:
            if space_storage is None:
                cls._space_storage_warning()

            hf_api.duplicate_space(
                from_id=_ARGILLA_SPACE_TEMPLATE_REPO,
                to_id=repo_id,
                private=private,
                exist_ok=True,
                hardware=space_hardware,
                storage=space_storage,
                secrets=secrets,
            )

        repo_url: RepoUrl = hf_api.create_repo(repo_id=repo_id, repo_type="space", exist_ok=True, space_sdk="docker")
        api_url: str = (
            f"https://{cls._sanitize_url_component(org_name)}-{cls._sanitize_url_component(repo_name)}.hf.space/"
        )
        cls._log_message(cls, message=f"Argilla is being deployed at: {repo_url}")
        while cls._is_building(hf_api.get_space_runtime(repo_id=repo_id).stage):
            time.sleep(_SLEEP_TIME)
            cls._log_message(cls, message=f"Deployment in progress. Waiting {_SLEEP_TIME} seconds.")

        headers = {}
        if private:
            headers["Authorization"] = f"Bearer {hf_token}"

        return cls(api_url=api_url, api_key=api_key, headers=headers)

    @staticmethod
    def _space_storage_warning() -> None:
        warnings.warn(
            "No storage provided. The space will not have persistent storage so every 48 hours your data will be reset."
        )

    @classmethod
    def _acquire_hf_token(cls, ht_token: Union[str, None]) -> str:
        """Obtain the Hugging Face authentication token to deploy a space and authenticate."""
        if ht_token is None:
            ht_token = get_token()
        if ht_token is None:
            if cls._is_interactive():
                notebook_login()
            else:
                login()
            ht_token = get_token()
        return ht_token

    @classmethod
    def _is_building(cls, stage: SpaceStage) -> bool:
        """Check the current stage of the space runtime. Simplified to return True when being built."""
        if stage in ["RUNNING_APP_STARTING", "RUNNING_BUILDING", "BUILDING", "APP_STARTING"]:
            return True
        elif stage in ["RUNNING", "PAUSED", "STOPPED"]:
            return False
        else:
            raise ValueError(f"Space configuration is wrong and in stage: {stage}")

    @classmethod
    def _is_space_stopped(cls, stage: SpaceStage) -> bool:
        """Check the current stage of the space runtime. Simplified to return True when it can be built."""
        if stage in ["RUNNING", "RUNNING_APP_STARTING", "RUNNING_BUILDING", "BUILDING", "APP_STARTING"]:
            return False
        elif stage in ["PAUSED", "STOPPED"]:
            return True
        else:
            raise ValueError(f"Space configuration is wrong and in stage: {stage}")

    @staticmethod
    def _sanitize_url_component(component: str) -> str:
        """Sanitize a component of a URL by replacing non-URL compatible characters."""

        # Replace any character that's not alphanumeric or hyphen with a hyphen
        sanitized = re.sub(r"[^a-zA-Z0-9-]", "-", component)
        # Convert to lowercase
        sanitized = sanitized.lower()
        # Remove any leading or trailing hyphens
        sanitized = sanitized.strip("-")
        return sanitized

deploy_on_spaces(api_key, repo_name='argilla', org_name=None, hf_token=None, space_storage=None, space_hardware='cpu-basic', private=False) classmethod

在 Hugging Face Spaces 上部署 Argilla。

参数

名称 类型 描述 默认值
api_key str

为 Space 的所有者用户和创建者定义的 Argilla API 密钥。

必需
repo_name 可选[str]

将在其中部署 Argilla 的存储库的 ID。 默认为 "argilla"。

'argilla'
org_name 可选[str]

将在其中部署 Argilla 的组织的名称。 默认为 None。

hf_token 可选[Union[str, None]]

Hugging Face 身份验证令牌。 默认为 None。

space_storage 可选[Union[str, SpaceStorage]]

Space 的持久存储大小。 默认为 None,不带持久存储。

space_hardware 可选[Union[str, SpaceHardware]]

Space 的硬件配置。 默认为 "cpu-basic",在不活动 48 小时后停机。

'cpu-basic'
private 可选[Union[bool, None]]

Space 是否应为私有。 默认为 False。

False

返回值

名称 类型 描述
Argilla Argilla

Argilla 客户端。

示例
import argilla as rg
client = rg.Argilla.deploy_on_spaces(api_key="12345678")
源代码位于 src/argilla/_helpers/_deploy.py
@classmethod
def deploy_on_spaces(
    cls,
    api_key: str,
    repo_name: Optional[str] = "argilla",
    org_name: Optional[str] = None,
    hf_token: Optional[str] = None,
    space_storage: Optional[Union[str, "SpaceStorage", Literal["small", "medium", "large"]]] = None,
    space_hardware: Optional[Union[str, "SpaceHardware", Literal["cpu-basic", "cpu-upgrade"]]] = "cpu-basic",
    private: Optional[Union[bool, None]] = False,
) -> "Argilla":
    """
    Deploys Argilla on Hugging Face Spaces.

    Args:
        api_key (str): The Argilla API key to be defined for the owner user and creator of the Space.
        repo_name (Optional[str]): The ID of the repository where Argilla will be deployed. Defaults to "argilla".
        org_name (Optional[str]): The name of the organization where Argilla will be deployed. Defaults to None.
        hf_token (Optional[Union[str, None]]): The Hugging Face authentication token. Defaults to None.
        space_storage (Optional[Union[str, SpaceStorage]]): The persistent storage size for the space. Defaults to None without persistent storage.
        space_hardware (Optional[Union[str, SpaceHardware]]): The hardware configuration for the space. Defaults to "cpu-basic" with downtime after 48 hours of inactivity.
        private (Optional[Union[bool, None]]): Whether the space should be private. Defaults to False.

    Returns:
        Argilla: The Argilla client.

    Example:
        ```Python
        import argilla as rg
        client = rg.Argilla.deploy_on_spaces(api_key="12345678")
        ```
    """
    hf_token = cls._acquire_hf_token(ht_token=hf_token)
    hf_api = HfApi(token=hf_token)

    # Get the org name from the repo name or default to the current user
    token_username = hf_api.whoami()["name"]
    org_name = org_name or token_username
    repo_id = f"{org_name}/{repo_name}"

    # Define the api_key for the space
    secrets = [
        {"key": "API_KEY", "value": api_key, "description": "The API key of the owner user."},
        {"key": "WORKSPACE", "value": "argilla", "description": "The workspace of the space."},
    ]

    # check API key length
    if len(api_key) < 8:
        raise ValueError(
            "Provided API key has invalid length. Please provide an apikey with at least 8 characters."
        )

    # Check if the space already exists
    if hf_api.repo_exists(repo_id=repo_id, repo_type="space"):
        if cls._is_space_stopped(hf_api.get_space_runtime(repo_id=repo_id).stage):
            hf_api.restart_space(repo_id=repo_id)
        warnings.warn(
            f"Space {repo_id} already exists. Using provided API key. If client authentication fails, go to "
            f"https://hugging-face.cn/spaces/{repo_id} to login with OAuth and get the correct API key.",
            stacklevel=2,
        )
    else:
        if space_storage is None:
            cls._space_storage_warning()

        hf_api.duplicate_space(
            from_id=_ARGILLA_SPACE_TEMPLATE_REPO,
            to_id=repo_id,
            private=private,
            exist_ok=True,
            hardware=space_hardware,
            storage=space_storage,
            secrets=secrets,
        )

    repo_url: RepoUrl = hf_api.create_repo(repo_id=repo_id, repo_type="space", exist_ok=True, space_sdk="docker")
    api_url: str = (
        f"https://{cls._sanitize_url_component(org_name)}-{cls._sanitize_url_component(repo_name)}.hf.space/"
    )
    cls._log_message(cls, message=f"Argilla is being deployed at: {repo_url}")
    while cls._is_building(hf_api.get_space_runtime(repo_id=repo_id).stage):
        time.sleep(_SLEEP_TIME)
        cls._log_message(cls, message=f"Deployment in progress. Waiting {_SLEEP_TIME} seconds.")

    headers = {}
    if private:
        headers["Authorization"] = f"Bearer {hf_token}"

    return cls(api_url=api_url, api_key=api_key, headers=headers)