Repository: apify/crawlee-python
File: src/crawlee/basic_crawler/_basic_crawler.py (Python)

Code before:
        dataset and then pushes the provided data to it.

        Args:
            data: The data to push to the dataset.
            dataset_id: The ID of the dataset.
            dataset_name: The name of the dataset.
            kwargs: Keyword arguments to be passed to the dataset's `push_data` method.
        """
        dataset = await self.get_dataset(id=dataset_id, name=dataset_name)
        await dataset.push_data(data, **kwargs)

    def _should_retry_request(self, crawling_context: BasicCrawlingContext, error: Exception) -> bool:
        if crawling_context.request.no_retry:
            return False

        # Do not retry on client errors.
        if isinstance(error, HttpStatusCodeError) and is_status_code_client_error(error.status_code):
            return False

        if isinstance(error, SessionError):
            return ((crawling_context.request.session_rotation_count or 0) + 1) < self._max_session_rotations

        max_request_retries = crawling_context.request.max_retries
        if max_request_retries is None:
            max_request_retries = self._max_request_retries

        return (crawling_context.request.retry_count + 1) < max_request_retries

    async def _check_url_after_redirects(
        self, crawling_context: TCrawlingContext
    ) -> AsyncGenerator[TCrawlingContext, None]:
        """Invoked at the end of the context pipeline to make sure that the `loaded_url` still matches enqueue_strategy.

        This is done to filter out links that redirect outside of the crawled domain.
        """
        if crawling_context.request.loaded_url is not None and not self._check_enqueue_strategy(
            crawling_context.request.enqueue_strategy,
            origin_url=urlparse(crawling_context.request.url),
            target_url=urlparse(crawling_context.request.loaded_url),
        ):
            raise ContextPipelineInterruptedError(
                f'Skipping URL {crawling_context.request.loaded_url} (redirected from {crawling_context.request.url})'
            )

        yield crawling_context

    def _check_enqueue_strategy(
        self,
        strategy: EnqueueStrategy,
        *,
        target_url: ParseResult,

Review comment (suggestion):
```suggestion
        path: str | Path,
        *,
```

Code after:
        """Export data from a dataset.

        This helper method simplifies the process of exporting data from a dataset. It opens the specified
        dataset and then exports the data based on the provided parameters. If you need to pass options
        specific to the output format, use the `export_data_csv` or `export_data_json` method instead.

        Args:
            path: The destination path.
            dataset_id: The ID of the dataset.
            dataset_name: The name of the dataset.
        """
        dataset = await self.get_dataset(id=dataset_id, name=dataset_name)

        path = path if isinstance(path, Path) else Path(path)
        destination = path.open('w', newline='')

        if path.suffix == '.csv':
            await dataset.write_to_csv(destination)
        elif path.suffix == '.json':
            await dataset.write_to_json(destination)
        else:
            raise ValueError(f'Unsupported file extension: {path.suffix}')

    async def export_data_csv(
        self,
        path: str | Path,
        *,
        dataset_id: str | None = None,
        dataset_name: str | None = None,
        **kwargs: Unpack[ExportDataCsvKwargs],
    ) -> None:
        """Export data from a dataset to a CSV file.

        This helper method simplifies the process of exporting data from a dataset in csv format. It opens the specified
        dataset and then exports the data based on the provided parameters.

        Args:
            path: The destination path.
            content_type: The output format.
            dataset_id: The ID of the dataset.
            dataset_name: The name of the dataset.
            kwargs: Extra configurations for dumping/writing in csv format.
        """
        dataset = await self.get_dataset(id=dataset_id, name=dataset_name)
        path = path if isinstance(path, Path) else Path(path)

        return await dataset.write_to_csv(path.open('w', newline=''), **kwargs)

    async def export_data_json(
        self,
        path: str | Path,