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

Code before:
        result = await self._http_client.crawl(
            request=context.request,
            session=context.session,
            proxy_info=context.proxy_info,
            statistics=self._statistics,
        )

        yield HttpCrawlingContext.from_basic_crawling_context(context=context, http_response=result.http_response)

    async def _handle_blocked_request(
        self, context: ParsedHttpCrawlingContext[TParseResult]
    ) -> AsyncGenerator[ParsedHttpCrawlingContext[TParseResult], None]:
        """Try to detect if the request is blocked based on the HTTP status code or the parsed response content.

        Args:
            context: The current crawling context.

        Raises:
            SessionError: If the request is considered blocked.

        Yields:
            The original crawling context if no errors are detected.
        """
        if self._retry_on_blocked:
            status_code = context.http_response.status_code
            if context.session and context.session.is_blocked_status_code(
                status_code=status_code,
                additional_blocked_status_codes=self._http_client.additional_blocked_status_codes,
                ignore_http_error_status_codes=self._http_client.ignore_http_error_status_codes,
            ):
                raise SessionError(f'Assuming the session is blocked based on HTTP status code {status_code}')
            if blocked_info := self._parser.is_blocked(context.parsed_content):
                raise SessionError(blocked_info.reason)
        yield context

    def pre_navigation_hook(self, hook: Callable[[BasicCrawlingContext], Awaitable[None]]) -> None:
        """Register a hook to be called before each navigation.

        Args:
            hook: A coroutine function to be called before each navigation.
        """
        self._pre_navigation_hooks.append(hook)


Review comment (refactor):
Did you check the equivalent method in `PlaywrightCrawler` as well? Any chance we could deduplicate this?

Code after:
        result = await self._http_client.crawl(
            request=context.request,
            session=context.session,
            proxy_info=context.proxy_info,
            statistics=self._statistics,
        )

        yield HttpCrawlingContext.from_basic_crawling_context(context=context, http_response=result.http_response)

    async def _handle_blocked_request(
        self, context: ParsedHttpCrawlingContext[TParseResult]
    ) -> AsyncGenerator[ParsedHttpCrawlingContext[TParseResult], None]:
        """Try to detect if the request is blocked based on the HTTP status code or the parsed response content.

        Args:
            context: The current crawling context.

        Raises:
            SessionError: If the request is considered blocked.

        Yields:
            The original crawling context if no errors are detected.
        """
        if self._retry_on_blocked:
            status_code = context.http_response.status_code
            if self._is_session_blocked_status_code(context.session, status_code):
                raise SessionError(f'Assuming the session is blocked based on HTTP status code {status_code}')
            if blocked_info := self._parser.is_blocked(context.parsed_content):
                raise SessionError(blocked_info.reason)
        yield context

    def pre_navigation_hook(self, hook: Callable[[BasicCrawlingContext], Awaitable[None]]) -> None:
        """Register a hook to be called before each navigation.

        Args:
            hook: A coroutine function to be called before each navigation.
        """
        self._pre_navigation_hooks.append(hook)