Argilla
zenml.integrations.argilla
special
Initialization of the Argilla integration.
ArgillaIntegration (Integration)
Definition of Argilla integration for ZenML.
Source code in zenml/integrations/argilla/__init__.py
class ArgillaIntegration(Integration):
"""Definition of Argilla integration for ZenML."""
NAME = ARGILLA
REQUIREMENTS = [
"argilla>=1.20.0,<2",
]
@classmethod
def flavors(cls) -> List[Type[Flavor]]:
"""Declare the stack component flavors for the Argilla integration.
Returns:
List of stack component flavors for this integration.
"""
from zenml.integrations.argilla.flavors import (
ArgillaAnnotatorFlavor,
)
return [ArgillaAnnotatorFlavor]
flavors()
classmethod
Declare the stack component flavors for the Argilla integration.
Returns:
Type | Description |
---|---|
List[Type[zenml.stack.flavor.Flavor]] |
List of stack component flavors for this integration. |
Source code in zenml/integrations/argilla/__init__.py
@classmethod
def flavors(cls) -> List[Type[Flavor]]:
"""Declare the stack component flavors for the Argilla integration.
Returns:
List of stack component flavors for this integration.
"""
from zenml.integrations.argilla.flavors import (
ArgillaAnnotatorFlavor,
)
return [ArgillaAnnotatorFlavor]
annotators
special
Initialization of the Argilla annotators submodule.
argilla_annotator
Implementation of the Argilla annotation integration.
ArgillaAnnotator (BaseAnnotator, AuthenticationMixin)
Class to interact with the Argilla annotation interface.
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
class ArgillaAnnotator(BaseAnnotator, AuthenticationMixin):
"""Class to interact with the Argilla annotation interface."""
@property
def config(self) -> ArgillaAnnotatorConfig:
"""Returns the `ArgillaAnnotatorConfig` config.
Returns:
The configuration.
"""
return cast(ArgillaAnnotatorConfig, self._config)
@property
def settings_class(self) -> Type[ArgillaAnnotatorSettings]:
"""Settings class for the Argilla annotator.
Returns:
The settings class.
"""
return ArgillaAnnotatorSettings
def get_url(self) -> str:
"""Gets the top-level URL of the annotation interface.
Returns:
The URL of the annotation interface.
"""
return (
f"{self.config.instance_url}:{self.config.port}"
if self.config.port
else self.config.instance_url
)
def _get_client(self) -> ArgillaClient:
"""Gets Argilla client.
Returns:
Argilla client.
"""
config = self.config
init_kwargs = {"api_url": self.get_url()}
# set the API key from the secret or using settings
authentication_secret = self.get_authentication_secret()
if config.api_key and authentication_secret:
api_key = config.api_key
logger.debug(
"Both API key and authentication secret are provided. Using API key from settings as priority."
)
elif authentication_secret:
api_key = authentication_secret.secret_values.get("api_key", "")
logger.debug("Using API key from secret.")
elif config.api_key is not None:
api_key = config.api_key
logger.debug("Using API key from settings.")
if api_key:
init_kwargs["api_key"] = api_key
if config.workspace is not None:
init_kwargs["workspace"] = config.workspace
if config.extra_headers is not None:
init_kwargs["extra_headers"] = json.loads(config.extra_headers)
if config.httpx_extra_kwargs is not None:
init_kwargs["httpx_extra_kwargs"] = json.loads(
config.httpx_extra_kwargs
)
try:
_ = rg.active_client()
except BaseClientError:
rg.init(**init_kwargs)
return rg.active_client()
def get_url_for_dataset(self, dataset_name: str) -> str:
"""Gets the URL of the annotation interface for the given dataset.
Args:
dataset_name: The name of the dataset.
Returns:
The URL of the annotation interface.
"""
dataset_id = self.get_dataset(dataset_name=dataset_name).id
return f"{self.get_url()}/dataset/{dataset_id}/annotation-mode"
def get_datasets(self) -> List[Any]:
"""Gets the datasets currently available for annotation.
Returns:
A list of datasets.
"""
old_datasets = self._get_client().list_datasets()
new_datasets = rg.FeedbackDataset.list()
# Deduplicate datasets based on their names
dataset_names = set()
deduplicated_datasets = []
for dataset in new_datasets + old_datasets:
if dataset.name not in dataset_names:
dataset_names.add(dataset.name)
deduplicated_datasets.append(dataset)
return deduplicated_datasets
def get_dataset_stats(self, dataset_name: str) -> Tuple[int, int]:
"""Gets the statistics of the given dataset.
Args:
dataset_name: The name of the dataset.
Returns:
A tuple containing (labeled_task_count, unlabeled_task_count) for
the dataset.
"""
dataset = self.get_dataset(dataset_name=dataset_name)
labeled_task_count = len(
dataset.filter_by(response_status="submitted")
)
unlabeled_task_count = len(
dataset.filter_by(response_status="pending")
)
return (labeled_task_count, unlabeled_task_count)
def add_dataset(self, **kwargs: Any) -> Any:
"""Registers a dataset for annotation.
You must pass a `dataset_name` and a `dataset` object to this method.
Args:
**kwargs: Additional keyword arguments to pass to the Argilla
client.
Returns:
An Argilla dataset object.
Raises:
ValueError: if 'dataset_name' and 'dataset' aren't provided.
"""
dataset_name = kwargs.get("dataset_name")
dataset = kwargs.get("dataset")
if not dataset_name:
raise ValueError("`dataset_name` keyword argument is required.")
elif dataset is None:
raise ValueError("`dataset` keyword argument is required.")
try:
logger.info(f"Pushing dataset '{dataset_name}' to Argilla...")
dataset.push_to_argilla(name=dataset_name)
logger.info(f"Dataset '{dataset_name}' pushed successfully.")
except Exception as e:
logger.error(
f"Failed to push dataset '{dataset_name}' to Argilla: {str(e)}"
)
raise ValueError(
f"Failed to push dataset to Argilla: {str(e)}"
) from e
return self.get_dataset(dataset_name=dataset_name)
def delete_dataset(self, **kwargs: Any) -> None:
"""Deletes a dataset from the annotation interface.
Args:
**kwargs: Additional keyword arguments to pass to the Argilla
client.
Raises:
ValueError: If the dataset name is not provided.
"""
dataset_name = kwargs.get("dataset_name")
if not dataset_name:
raise ValueError("`dataset_name` keyword argument is required.")
try:
self._get_client().delete(name=dataset_name)
self.get_dataset(dataset_name=dataset_name).delete()
logger.info(f"Dataset '{dataset_name}' deleted successfully.")
except ValueError:
logger.warning(
f"Dataset '{dataset_name}' not found. Skipping deletion."
)
def get_dataset(self, **kwargs: Any) -> Any:
"""Gets the dataset with the given name.
Args:
**kwargs: Additional keyword arguments to pass to the Argilla client.
Returns:
The Argilla DatasetModel object for the given name.
Raises:
ValueError: If the dataset name is not provided or if the dataset
does not exist.
"""
dataset_name = kwargs.get("dataset_name")
if not dataset_name:
raise ValueError("`dataset_name` keyword argument is required.")
try:
if rg.FeedbackDataset.from_argilla(name=dataset_name) is not None:
return rg.FeedbackDataset.from_argilla(name=dataset_name)
else:
return self._get_client().get_dataset(name=dataset_name)
except (NotFoundApiError, ValueError) as e:
logger.error(f"Dataset '{dataset_name}' not found.")
raise ValueError(f"Dataset '{dataset_name}' not found.") from e
def get_data_by_status(self, dataset_name: str, status: str) -> Any:
"""Gets the dataset containing the data with the specified status.
Args:
dataset_name: The name of the dataset.
status: The response status to filter by ('submitted' for labeled,
'pending' for unlabeled).
Returns:
The dataset containing the data with the specified status.
Raises:
ValueError: If the dataset name is not provided.
"""
if not dataset_name:
raise ValueError("`dataset_name` argument is required.")
return self.get_dataset(dataset_name=dataset_name).filter_by(
response_status=status
)
def get_labeled_data(self, **kwargs: Any) -> Any:
"""Gets the dataset containing the labeled data.
Args:
**kwargs: Additional keyword arguments to pass to the Argilla client.
Returns:
The dataset containing the labeled data.
Raises:
ValueError: If the dataset name is not provided.
"""
if dataset_name := kwargs.get("dataset_name"):
return self.get_data_by_status(dataset_name, status="submitted")
else:
raise ValueError("`dataset_name` keyword argument is required.")
def get_unlabeled_data(self, **kwargs: str) -> Any:
"""Gets the dataset containing the unlabeled data.
Args:
**kwargs: Additional keyword arguments to pass to the Argilla client.
Returns:
The dataset containing the unlabeled data.
Raises:
ValueError: If the dataset name is not provided.
"""
if dataset_name := kwargs.get("dataset_name"):
return self.get_data_by_status(dataset_name, status="pending")
else:
raise ValueError("`dataset_name` keyword argument is required.")
config: ArgillaAnnotatorConfig
property
readonly
Returns the ArgillaAnnotatorConfig
config.
Returns:
Type | Description |
---|---|
ArgillaAnnotatorConfig |
The configuration. |
settings_class: Type[zenml.integrations.argilla.flavors.argilla_annotator_flavor.ArgillaAnnotatorSettings]
property
readonly
Settings class for the Argilla annotator.
Returns:
Type | Description |
---|---|
Type[zenml.integrations.argilla.flavors.argilla_annotator_flavor.ArgillaAnnotatorSettings] |
The settings class. |
add_dataset(self, **kwargs)
Registers a dataset for annotation.
You must pass a dataset_name
and a dataset
object to this method.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
**kwargs |
Any |
Additional keyword arguments to pass to the Argilla client. |
{} |
Returns:
Type | Description |
---|---|
Any |
An Argilla dataset object. |
Exceptions:
Type | Description |
---|---|
ValueError |
if 'dataset_name' and 'dataset' aren't provided. |
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
def add_dataset(self, **kwargs: Any) -> Any:
"""Registers a dataset for annotation.
You must pass a `dataset_name` and a `dataset` object to this method.
Args:
**kwargs: Additional keyword arguments to pass to the Argilla
client.
Returns:
An Argilla dataset object.
Raises:
ValueError: if 'dataset_name' and 'dataset' aren't provided.
"""
dataset_name = kwargs.get("dataset_name")
dataset = kwargs.get("dataset")
if not dataset_name:
raise ValueError("`dataset_name` keyword argument is required.")
elif dataset is None:
raise ValueError("`dataset` keyword argument is required.")
try:
logger.info(f"Pushing dataset '{dataset_name}' to Argilla...")
dataset.push_to_argilla(name=dataset_name)
logger.info(f"Dataset '{dataset_name}' pushed successfully.")
except Exception as e:
logger.error(
f"Failed to push dataset '{dataset_name}' to Argilla: {str(e)}"
)
raise ValueError(
f"Failed to push dataset to Argilla: {str(e)}"
) from e
return self.get_dataset(dataset_name=dataset_name)
delete_dataset(self, **kwargs)
Deletes a dataset from the annotation interface.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
**kwargs |
Any |
Additional keyword arguments to pass to the Argilla client. |
{} |
Exceptions:
Type | Description |
---|---|
ValueError |
If the dataset name is not provided. |
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
def delete_dataset(self, **kwargs: Any) -> None:
"""Deletes a dataset from the annotation interface.
Args:
**kwargs: Additional keyword arguments to pass to the Argilla
client.
Raises:
ValueError: If the dataset name is not provided.
"""
dataset_name = kwargs.get("dataset_name")
if not dataset_name:
raise ValueError("`dataset_name` keyword argument is required.")
try:
self._get_client().delete(name=dataset_name)
self.get_dataset(dataset_name=dataset_name).delete()
logger.info(f"Dataset '{dataset_name}' deleted successfully.")
except ValueError:
logger.warning(
f"Dataset '{dataset_name}' not found. Skipping deletion."
)
get_data_by_status(self, dataset_name, status)
Gets the dataset containing the data with the specified status.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
dataset_name |
str |
The name of the dataset. |
required |
status |
str |
The response status to filter by ('submitted' for labeled, 'pending' for unlabeled). |
required |
Returns:
Type | Description |
---|---|
Any |
The dataset containing the data with the specified status. |
Exceptions:
Type | Description |
---|---|
ValueError |
If the dataset name is not provided. |
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
def get_data_by_status(self, dataset_name: str, status: str) -> Any:
"""Gets the dataset containing the data with the specified status.
Args:
dataset_name: The name of the dataset.
status: The response status to filter by ('submitted' for labeled,
'pending' for unlabeled).
Returns:
The dataset containing the data with the specified status.
Raises:
ValueError: If the dataset name is not provided.
"""
if not dataset_name:
raise ValueError("`dataset_name` argument is required.")
return self.get_dataset(dataset_name=dataset_name).filter_by(
response_status=status
)
get_dataset(self, **kwargs)
Gets the dataset with the given name.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
**kwargs |
Any |
Additional keyword arguments to pass to the Argilla client. |
{} |
Returns:
Type | Description |
---|---|
Any |
The Argilla DatasetModel object for the given name. |
Exceptions:
Type | Description |
---|---|
ValueError |
If the dataset name is not provided or if the dataset does not exist. |
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
def get_dataset(self, **kwargs: Any) -> Any:
"""Gets the dataset with the given name.
Args:
**kwargs: Additional keyword arguments to pass to the Argilla client.
Returns:
The Argilla DatasetModel object for the given name.
Raises:
ValueError: If the dataset name is not provided or if the dataset
does not exist.
"""
dataset_name = kwargs.get("dataset_name")
if not dataset_name:
raise ValueError("`dataset_name` keyword argument is required.")
try:
if rg.FeedbackDataset.from_argilla(name=dataset_name) is not None:
return rg.FeedbackDataset.from_argilla(name=dataset_name)
else:
return self._get_client().get_dataset(name=dataset_name)
except (NotFoundApiError, ValueError) as e:
logger.error(f"Dataset '{dataset_name}' not found.")
raise ValueError(f"Dataset '{dataset_name}' not found.") from e
get_dataset_stats(self, dataset_name)
Gets the statistics of the given dataset.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
dataset_name |
str |
The name of the dataset. |
required |
Returns:
Type | Description |
---|---|
Tuple[int, int] |
A tuple containing (labeled_task_count, unlabeled_task_count) for the dataset. |
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
def get_dataset_stats(self, dataset_name: str) -> Tuple[int, int]:
"""Gets the statistics of the given dataset.
Args:
dataset_name: The name of the dataset.
Returns:
A tuple containing (labeled_task_count, unlabeled_task_count) for
the dataset.
"""
dataset = self.get_dataset(dataset_name=dataset_name)
labeled_task_count = len(
dataset.filter_by(response_status="submitted")
)
unlabeled_task_count = len(
dataset.filter_by(response_status="pending")
)
return (labeled_task_count, unlabeled_task_count)
get_datasets(self)
Gets the datasets currently available for annotation.
Returns:
Type | Description |
---|---|
List[Any] |
A list of datasets. |
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
def get_datasets(self) -> List[Any]:
"""Gets the datasets currently available for annotation.
Returns:
A list of datasets.
"""
old_datasets = self._get_client().list_datasets()
new_datasets = rg.FeedbackDataset.list()
# Deduplicate datasets based on their names
dataset_names = set()
deduplicated_datasets = []
for dataset in new_datasets + old_datasets:
if dataset.name not in dataset_names:
dataset_names.add(dataset.name)
deduplicated_datasets.append(dataset)
return deduplicated_datasets
get_labeled_data(self, **kwargs)
Gets the dataset containing the labeled data.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
**kwargs |
Any |
Additional keyword arguments to pass to the Argilla client. |
{} |
Returns:
Type | Description |
---|---|
Any |
The dataset containing the labeled data. |
Exceptions:
Type | Description |
---|---|
ValueError |
If the dataset name is not provided. |
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
def get_labeled_data(self, **kwargs: Any) -> Any:
"""Gets the dataset containing the labeled data.
Args:
**kwargs: Additional keyword arguments to pass to the Argilla client.
Returns:
The dataset containing the labeled data.
Raises:
ValueError: If the dataset name is not provided.
"""
if dataset_name := kwargs.get("dataset_name"):
return self.get_data_by_status(dataset_name, status="submitted")
else:
raise ValueError("`dataset_name` keyword argument is required.")
get_unlabeled_data(self, **kwargs)
Gets the dataset containing the unlabeled data.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
**kwargs |
str |
Additional keyword arguments to pass to the Argilla client. |
{} |
Returns:
Type | Description |
---|---|
Any |
The dataset containing the unlabeled data. |
Exceptions:
Type | Description |
---|---|
ValueError |
If the dataset name is not provided. |
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
def get_unlabeled_data(self, **kwargs: str) -> Any:
"""Gets the dataset containing the unlabeled data.
Args:
**kwargs: Additional keyword arguments to pass to the Argilla client.
Returns:
The dataset containing the unlabeled data.
Raises:
ValueError: If the dataset name is not provided.
"""
if dataset_name := kwargs.get("dataset_name"):
return self.get_data_by_status(dataset_name, status="pending")
else:
raise ValueError("`dataset_name` keyword argument is required.")
get_url(self)
Gets the top-level URL of the annotation interface.
Returns:
Type | Description |
---|---|
str |
The URL of the annotation interface. |
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
def get_url(self) -> str:
"""Gets the top-level URL of the annotation interface.
Returns:
The URL of the annotation interface.
"""
return (
f"{self.config.instance_url}:{self.config.port}"
if self.config.port
else self.config.instance_url
)
get_url_for_dataset(self, dataset_name)
Gets the URL of the annotation interface for the given dataset.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
dataset_name |
str |
The name of the dataset. |
required |
Returns:
Type | Description |
---|---|
str |
The URL of the annotation interface. |
Source code in zenml/integrations/argilla/annotators/argilla_annotator.py
def get_url_for_dataset(self, dataset_name: str) -> str:
"""Gets the URL of the annotation interface for the given dataset.
Args:
dataset_name: The name of the dataset.
Returns:
The URL of the annotation interface.
"""
dataset_id = self.get_dataset(dataset_name=dataset_name).id
return f"{self.get_url()}/dataset/{dataset_id}/annotation-mode"
flavors
special
Argilla integration flavors.
argilla_annotator_flavor
Argilla annotator flavor.
ArgillaAnnotatorConfig (BaseAnnotatorConfig, ArgillaAnnotatorSettings, AuthenticationConfigMixin)
Config for the Argilla annotator.
This class combines settings and authentication configurations for Argilla into a single, usable configuration object without adding additional functionality.
Source code in zenml/integrations/argilla/flavors/argilla_annotator_flavor.py
class ArgillaAnnotatorConfig(
BaseAnnotatorConfig,
ArgillaAnnotatorSettings,
AuthenticationConfigMixin,
):
"""Config for the Argilla annotator.
This class combines settings and authentication configurations for
Argilla into a single, usable configuration object without adding
additional functionality.
"""
ArgillaAnnotatorFlavor (BaseAnnotatorFlavor)
Argilla annotator flavor.
Source code in zenml/integrations/argilla/flavors/argilla_annotator_flavor.py
class ArgillaAnnotatorFlavor(BaseAnnotatorFlavor):
"""Argilla annotator flavor."""
@property
def name(self) -> str:
"""Name of the flavor.
Returns:
The name of the flavor.
"""
return ARGILLA_ANNOTATOR_FLAVOR
@property
def docs_url(self) -> Optional[str]:
"""A url to point at docs explaining this flavor.
Returns:
A flavor docs url.
"""
return self.generate_default_docs_url()
@property
def sdk_docs_url(self) -> Optional[str]:
"""A url to point at SDK docs explaining this flavor.
Returns:
A flavor SDK docs url.
"""
return self.generate_default_sdk_docs_url()
@property
def logo_url(self) -> str:
"""A url to represent the flavor in the dashboard.
Returns:
The flavor logo.
"""
return "https://public-flavor-logos.s3.eu-central-1.amazonaws.com/annotator/argilla.png"
@property
def config_class(self) -> Type[ArgillaAnnotatorConfig]:
"""Returns `ArgillaAnnotatorConfig` config class.
Returns:
The config class.
"""
return ArgillaAnnotatorConfig
@property
def implementation_class(self) -> Type["ArgillaAnnotator"]:
"""Implementation class for this flavor.
Returns:
The implementation class.
"""
from zenml.integrations.argilla.annotators import (
ArgillaAnnotator,
)
return ArgillaAnnotator
config_class: Type[zenml.integrations.argilla.flavors.argilla_annotator_flavor.ArgillaAnnotatorConfig]
property
readonly
Returns ArgillaAnnotatorConfig
config class.
Returns:
Type | Description |
---|---|
Type[zenml.integrations.argilla.flavors.argilla_annotator_flavor.ArgillaAnnotatorConfig] |
The config class. |
docs_url: Optional[str]
property
readonly
A url to point at docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor docs url. |
implementation_class: Type[ArgillaAnnotator]
property
readonly
Implementation class for this flavor.
Returns:
Type | Description |
---|---|
Type[ArgillaAnnotator] |
The implementation class. |
logo_url: str
property
readonly
A url to represent the flavor in the dashboard.
Returns:
Type | Description |
---|---|
str |
The flavor logo. |
name: str
property
readonly
Name of the flavor.
Returns:
Type | Description |
---|---|
str |
The name of the flavor. |
sdk_docs_url: Optional[str]
property
readonly
A url to point at SDK docs explaining this flavor.
Returns:
Type | Description |
---|---|
Optional[str] |
A flavor SDK docs url. |
ArgillaAnnotatorSettings (BaseSettings)
Argilla annotator settings.
If you are using a private Hugging Face Spaces instance of Argilla you must pass in https_extra_kwargs.
Attributes:
Name | Type | Description |
---|---|---|
instance_url |
str |
URL of the Argilla instance. |
api_key |
Optional[str] |
The api_key for Argilla |
workspace |
Optional[str] |
The workspace to use for the annotation interface. |
port |
Optional[int] |
The port to use for the annotation interface. |
extra_headers |
Optional[str] |
Extra headers to include in the request. |
httpx_extra_kwargs |
Optional[str] |
Extra kwargs to pass to the client. |
Source code in zenml/integrations/argilla/flavors/argilla_annotator_flavor.py
class ArgillaAnnotatorSettings(BaseSettings):
"""Argilla annotator settings.
If you are using a private Hugging Face Spaces instance of Argilla you
must pass in https_extra_kwargs.
Attributes:
instance_url: URL of the Argilla instance.
api_key: The api_key for Argilla
workspace: The workspace to use for the annotation interface.
port: The port to use for the annotation interface.
extra_headers: Extra headers to include in the request.
httpx_extra_kwargs: Extra kwargs to pass to the client.
"""
instance_url: str = DEFAULT_LOCAL_INSTANCE_URL
api_key: Optional[str] = SecretField(default=None)
workspace: Optional[str] = "admin"
port: Optional[int]
extra_headers: Optional[str] = None
httpx_extra_kwargs: Optional[str] = None
@field_validator("instance_url")
@classmethod
def ensure_instance_url_ends_without_slash(cls, instance_url: str) -> str:
"""Pydantic validator to ensure instance URL ends without a slash.
Args:
instance_url: The instance URL to validate.
Returns:
The validated instance URL.
"""
return instance_url.rstrip("/")
ensure_instance_url_ends_without_slash(instance_url)
classmethod
Pydantic validator to ensure instance URL ends without a slash.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
instance_url |
str |
The instance URL to validate. |
required |
Returns:
Type | Description |
---|---|
str |
The validated instance URL. |
Source code in zenml/integrations/argilla/flavors/argilla_annotator_flavor.py
@field_validator("instance_url")
@classmethod
def ensure_instance_url_ends_without_slash(cls, instance_url: str) -> str:
"""Pydantic validator to ensure instance URL ends without a slash.
Args:
instance_url: The instance URL to validate.
Returns:
The validated instance URL.
"""
return instance_url.rstrip("/")