Skip to content

rg.Argilla

To interact with the Argilla server from Python you can use the Argilla class. The Argilla client is used to create, get, update, and delete all Argilla resources, such as workspaces, users, datasets, and records.

Usage Examples

Deploying Argilla Server on Hugging Face Spaces

To deploy Argilla on Hugging Face Spaces, use the deploy_on_spaces method.

import argilla as rg

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

Connecting to an Argilla server

To connect to an Argilla server, instantiate the Argilla class and pass the api_url of the server and the api_key to authenticate.

import argilla as rg

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

Accessing Dataset, Workspace, and User objects

The Argilla clients provides access to the Dataset, Workspace, and User objects of the Argilla server.

my_dataset = client.datasets("my_dataset")

my_workspace = client.workspaces("my_workspace")

my_user = client.users("my_user")

These resources can then be interacted with to access their properties and methods. For example, to list all datasets in a workspace:

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

Argilla

Bases: APIClient, SpacesDeploymentMixin, NotebookHTMLReprMixin

Argilla API client. This is the main entry point to interact with the API.

Attributes:

Name Type Description
workspaces Workspaces

A collection of workspaces.
datasets Datasets

A collection of datasets.
users Users

A collection of users.
me User

The current user.
Source code in src/argilla/client.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
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)

    @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

A collection of workspaces on the server.

datasets: Datasets property

A collection of datasets on the server.

users: Users property

A collection of users on the server.

__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)

Inits the Argilla client.

Parameters:

Name Type Description Default
api_url Optional[str]

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_url
api_key Optional[str]

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.

 api_key
timeout int

the maximum time in seconds to wait for a request to the Argilla API to be completed before raising an exception. Defaults to 60.

 timeout
retries int

the number of times to retry the HTTP connection to the Argilla API before raising an exception. Defaults to 5.

 retries
Source code in src/argilla/client.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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

Bases: Sequence['User'], ResourceHTMLReprMixin

A collection of users. It can be used to create a new user or to get an existing one.

Source code in src/argilla/client.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
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)

Add a new user to Argilla.

Parameters:

Name Type Description Default
user User

User object.

 required

Returns:

Name Type Description
User User

The created user.
Source code in src/argilla/client.py 
170
171
172
173
174
175
176
177
178
179
180
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)

List all users.

Source code in src/argilla/client.py 
188
189
190
191
192
193
194
195
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

Bases: Sequence['Workspace'], ResourceHTMLReprMixin

A collection of workspaces. It can be used to create a new workspace or to get an existing one.

Source code in src/argilla/client.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
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

The default workspace.

add(workspace)

Add a new workspace to the Argilla platform. Args: workspace: Workspace object.

Returns:

Name Type Description
Workspace Workspace

The created workspace.
Source code in src/argilla/client.py 
267
268
269
270
271
272
273
274
275
276
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

Bases: Sequence['Dataset'], ResourceHTMLReprMixin

A collection of datasets. It can be used to create a new dataset or to get an existing one.

Source code in src/argilla/client.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
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"]):
        pass

    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.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)

    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 [self._from_model(model) for model in self._api.list()]

    ############################
    # 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)

Add a new dataset to the Argilla platform

Parameters:

Name Type Description Default
dataset Dataset

Dataset object.

 required

Returns:

Name Type Description
Dataset Dataset

The created dataset.
Source code in src/argilla/client.py 
368
369
370
371
372
373
374
375
376
377
378
379
380
381
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

Bases: LoggingMixin

Source code in src/argilla/_helpers/_deploy.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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 persistant storage size for the space. Defaults to None without persistant 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
        api
                    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://huggingface.co/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 persistant 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 

    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 persistant storage size for the space. Defaults to None without persistant 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

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

Source code in src/argilla/_helpers/_deploy.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@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 persistant storage size for the space. Defaults to None without persistant 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
    api
                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://huggingface.co/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)