def log(
        self,
        records: Union[Record, Iterable[Record], Dataset],
        name: str,
        workspace: Optional[str] = None,
        tags: Optional[Dict[str, str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        batch_size: int = 100,
        verbose: bool = True,
        background: bool = False,
        num_threads: int = 0,
        max_retries: int = 3,
        chunk_size: Optional[int] = None,
    ) -> Union[BulkResponse, Future]:
        """Logs Records to argilla.

        The logging happens asynchronously in a background thread.

        Args:
            records: The record, an iterable of records, or a dataset to log.
            name: The dataset name.
            tags: A dictionary of tags related to the dataset.
            metadata: A dictionary of extra info for the dataset.
            batch_size: The batch size for a data bulk.
            verbose: If True, shows a progress bar and prints out a quick summary at the end.
            background: If True, we will NOT wait for the logging process to finish and return
                an ``asyncio.Future`` object. You probably want to set ``verbose`` to False
                in that case.
            num_threads: If > 0, will use num_thread separate number threads to batches, sending data concurrently.
                Default to `0`, which means no threading at all.
            max_retries: Number of retries when logging a batch of records if a `httpx.TransportError` occurs.
                Default `3`
            chunk_size: DEPRECATED! Use `batch_size` instead.

        Returns:
            Summary of the response from the REST API.
            If the ``background`` argument is set to True, an ``asyncio.Future``
            will be returned instead.

        """

        if background:
            executor = ThreadPoolExecutor(max_workers=1)

            return executor.submit(
                self.log,
                records=records,
                name=name,
                workspace=workspace,
                tags=tags,
                metadata=metadata,
                batch_size=batch_size,
                verbose=verbose,
                chunk_size=chunk_size,
                num_threads=num_threads,
                max_retries=max_retries,
            )

        tags = tags or {}
        metadata = metadata or {}

        if workspace is not None:
            self.set_workspace(workspace)

        if not name:
            raise InputValueError("Empty dataset name has been passed as argument.")

        if not re.match(DATASET_NAME_REGEX_PATTERN, name):
            raise InputValueError(
                f"Provided dataset name {name} does not match the pattern"
                f" {DATASET_NAME_REGEX_PATTERN}. Please, use a valid name for your"
                " dataset. This limitation is caused by naming conventions for indexes"
                " in Elasticsearch."
                " https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-create-index.html"
            )

        if chunk_size is not None:
            warnings.warn(
                "The argument `chunk_size` is deprecated and will be removed in a future"
                " version. Please use `batch_size` instead.",
                FutureWarning,
            )
            batch_size = chunk_size

        if batch_size > self._MAX_BATCH_SIZE:
            warnings.warn(
                "The requested batch size is noticeably large, timeout errors may occur. "
                f"Consider a batch size smaller than {self._MAX_BATCH_SIZE}",
            )

        if isinstance(records, Record.__args__):
            records = [records]
        records = list(records)

        try:
            record_type = type(records[0])
        except IndexError:
            raise InputValueError("Empty record list has been passed as argument.")

        if record_type is TextClassificationRecord:
            bulk_class = TextClassificationBulkData
            creation_class = CreationTextClassificationRecord
            task = TaskType.text_classification
        elif record_type is TokenClassificationRecord:
            bulk_class = TokenClassificationBulkData
            creation_class = CreationTokenClassificationRecord
            task = TaskType.token_classification
        elif record_type is Text2TextRecord:
            bulk_class = Text2TextBulkData
            creation_class = CreationText2TextRecord
            task = TaskType.text2text
        else:
            raise InputValueError(f"Unknown record type {record_type}. Available values are {Record.__args__}")

        try:
            self.datasets.create(name=name, task=task, workspace=workspace)
        except AlreadyExistsApiError:
            pass

        results = []
        with Progress() as progress_bar:
            task = progress_bar.add_task("Logging...", total=len(records), visible=verbose)

            batches = [records[i : i + batch_size] for i in range(0, len(records), batch_size)]

            @backoff.on_exception(
                backoff.expo,
                exception=httpx.TransportError,
                max_tries=max_retries,
                backoff_log_level=logging.DEBUG,
            )
            def log_batch(batch_info: Tuple[int, list]) -> Tuple[int, int]:
                batch_id, batch = batch_info

                bulk_result = bulk(
                    client=self.http_client,
                    name=name,
                    json_body=bulk_class(
                        tags=tags, metadata=metadata, records=[creation_class.from_client(r) for r in batch]
                    ),
                )

                progress_bar.update(task, advance=len(batch))
                return bulk_result.processed, bulk_result.failed

            if num_threads >= 1:
                with ThreadPoolExecutor(max_workers=num_threads) as executor:
                    results.extend(list(executor.map(log_batch, enumerate(batches))))
            else:
                results.extend(list(map(log_batch, enumerate(batches))))

        processed_batches, failed_batches = zip(*results)
        processed, failed = sum(processed_batches), sum(failed_batches)

        # TODO: improve logging policy in library
        if verbose:
            _LOGGER.info(f"Processed {processed} records in dataset {name}. Failed: {failed}")
            workspace = self.get_workspace()
            if not workspace:  # Just for backward comp. with datasets with no workspaces
                workspace = "-"
            url = f"{self.http_client.base_url}/datasets/{workspace}/{name}"
            rprint(f"{processed} records logged to [link={url}]{url}[/link]")

        # Creating a composite BulkResponse with the total processed and failed
        return BulkResponse(dataset=name, processed=processed, failed=failed)
Logs Records to argilla.

        The logging happens asynchronously in a background thread.

        Args:
            records: The record, an iterable of records, or a dataset to log.
            name: The dataset name.
            tags: A dictionary of tags related to the dataset.
            metadata: A dictionary of extra info for the dataset.
            batch_size: The batch size for a data bulk.
            verbose: If True, shows a progress bar and prints out a quick summary at the end.
            background: If True, we will NOT wait for the logging process to finish and return
                an ``asyncio.Future`` object. You probably want to set ``verbose`` to False
                in that case.
            num_threads: If > 0, will use num_thread separate number threads to batches, sending data concurrently.
                Default to `0`, which means no threading at all.
            max_retries: Number of retries when logging a batch of records if a `httpx.TransportError` occurs.
                Default `3`
            chunk_size: DEPRECATED! Use `batch_size` instead.

        Returns:
            Summary of the response from the REST API.
            If the ``background`` argument is set to True, an ``asyncio.Future``
            will be returned instead.

        
argilla-v1/src/argilla_v1/client/client.py