Repository: apify/crawlee-python
File: src/crawlee/crawlers/_abstract_http/_abstract_http_crawler.py (Python)

Code before:
        parsed_content = await self._parser.parse(context.http_response)
        yield ParsedHttpCrawlingContext.from_http_crawling_context(
            context=context,
            parsed_content=parsed_content,
            enqueue_links=self._create_enqueue_links_function(context, parsed_content),
        )

    def _create_enqueue_links_function(
        self, context: HttpCrawlingContext, parsed_content: TParseResult
    ) -> EnqueueLinksFunction:
        """Create a callback function for extracting links from parsed content and enqueuing them to the crawl.

        Args:
            context: The current crawling context.
            parsed_content: The parsed http response.

        Returns:
            Awaitable that is used for extracting links from parsed content and enqueuing them to the crawl.
        """

        async def enqueue_links(
            *,
            selector: str = 'a',
            label: str | None = None,
            user_data: dict[str, Any] | None = None,
            transform_request_function: Callable[[RequestOptions], RequestOptions | Literal['skip', 'unchanged']]
            | None = None,
            **kwargs: Unpack[EnqueueLinksKwargs],
        ) -> None:
            kwargs.setdefault('strategy', EnqueueStrategy.SAME_HOSTNAME)

            requests = list[Request]()
            base_user_data = user_data or {}

            for link in self._parser.find_links(parsed_content, selector=selector):
                url = link
                if not is_url_absolute(url):
                    url = convert_to_absolute_url(context.request.url, url)

                request_option = RequestOptions(url=url, user_data={**base_user_data}, label=label)

                if transform_request_function:
                    transform_request_option = transform_request_function(request_option)
                    if transform_request_option == 'skip':
                        continue
                    if transform_request_option != 'unchanged':
                        request_option = transform_request_option

                try:
                    request = Request.from_url(**request_option)
                except ValidationError as exc:

Review comment (refactor):
We should extract this type into some shared module. Users who want to provide a correct return type annotation for their `transform_request_function` should not need to copy all this.

Code after:
        parsed_content = await self._parser.parse(context.http_response)
        yield ParsedHttpCrawlingContext.from_http_crawling_context(
            context=context,
            parsed_content=parsed_content,
            enqueue_links=self._create_enqueue_links_function(context, parsed_content),
        )

    def _create_enqueue_links_function(
        self, context: HttpCrawlingContext, parsed_content: TParseResult
    ) -> EnqueueLinksFunction:
        """Create a callback function for extracting links from parsed content and enqueuing them to the crawl.

        Args:
            context: The current crawling context.
            parsed_content: The parsed http response.

        Returns:
            Awaitable that is used for extracting links from parsed content and enqueuing them to the crawl.
        """

        async def enqueue_links(
            *,
            selector: str = 'a',
            label: str | None = None,
            user_data: dict[str, Any] | None = None,
            transform_request_function: Callable[[RequestOptions], RequestOptions | RequestTransformAction]
            | None = None,
            **kwargs: Unpack[EnqueueLinksKwargs],
        ) -> None:
            kwargs.setdefault('strategy', EnqueueStrategy.SAME_HOSTNAME)

            requests = list[Request]()
            base_user_data = user_data or {}

            for link in self._parser.find_links(parsed_content, selector=selector):
                url = link
                if not is_url_absolute(url):
                    url = convert_to_absolute_url(context.request.url, url)

                request_options = RequestOptions(url=url, user_data={**base_user_data}, label=label)

                if transform_request_function:
                    transform_request_options = transform_request_function(request_options)
                    if transform_request_options == 'skip':
                        continue
                    if transform_request_options != 'unchanged':
                        request_options = transform_request_options

                try:
                    request = Request.from_url(**request_options)
                except ValidationError as exc: