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("/")