ronantakizawa/codereview-bench huggingface dataset:
Repository: apify/crawlee-python
File: docs/examples/code_examples/resuming_paused_crawl.py (Python)

Code before:
import asyncio

from crawlee import ConcurrencySettings, service_locator
from crawlee.crawlers import (
    BeautifulSoupCrawler,
    BeautifulSoupCrawlingContext,
)

# Disable clearing the `Queue` and `Dataset` on each run
# This allows you to continue from where you left off in the previous run
# We recommend using the environment variable `CRAWLEE_PURGE_ON_START=0`
configuration = service_locator.get_configuration()
configuration.purge_on_start = False


async def main() -> None:
    crawler = BeautifulSoupCrawler(
        # Let's slow down the crawler for a demonstration
        concurrency_settings=ConcurrencySettings(max_tasks_per_minute=20)
    )

    @crawler.router.default_handler
    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
        context.log.info(f'Processing {context.request.url} ...')

    # List of links for crawl
    requests = [
        'https://crawlee.dev',
        'https://crawlee.dev/python/docs',
        'https://crawlee.dev/python/docs/examples',
        'https://crawlee.dev/python/docs/guides',
        'https://crawlee.dev/python/docs/quick-start',
    ]

    await crawler.run(requests)


Review comment (suggestion):
```suggestion
# Disable clearing the `RequestQueue`, `KeyValueStore` and `Dataset` on each run.
# This makes the scraper continue from where it left off in the previous run.
# The recommended way to achieve this behavior is setting the environment variable `CRAWLEE_PURGE_ON_START=0`
```

Code after:
import asyncio

from crawlee import ConcurrencySettings, service_locator
from crawlee.crawlers import (
    BeautifulSoupCrawler,
    BeautifulSoupCrawlingContext,
)

# Disable clearing the `RequestQueue`, `KeyValueStore` and `Dataset` on each run.
# This makes the scraper continue from where it left off in the previous run.
# The recommended way to achieve this behavior is setting the environment variable
# `CRAWLEE_PURGE_ON_START=0`
configuration = service_locator.get_configuration()
configuration.purge_on_start = False


async def main() -> None:
    crawler = BeautifulSoupCrawler(
        # Let's slow down the crawler for a demonstration
        concurrency_settings=ConcurrencySettings(max_tasks_per_minute=20)
    )

    @crawler.router.default_handler
    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
        context.log.info(f'Processing {context.request.url} ...')

    # List of links for crawl
    requests = [
        'https://crawlee.dev',
        'https://crawlee.dev/python/docs',
        'https://crawlee.dev/python/docs/examples',
        'https://crawlee.dev/python/docs/guides',
        'https://crawlee.dev/python/docs/quick-start',
    ]

    await crawler.run(requests)
Repository: apify/crawlee-python
File: tests/unit/crawlers/_playwright/test_playwright_crawler.py (Python)

Code before:

async def test_send_request_with_client(server_url: URL) -> None:
    """Check that the persist context works with fingerprints."""
    check_data: dict[str, Any] = {}

    crawler = PlaywrightCrawler(
        http_client=HttpxHttpClient(header_generator=None, headers={'user-agent': 'My User-Agent'})
    )

    @crawler.router.default_handler
    async def request_handler(context: PlaywrightCrawlingContext) -> None:
        response = await context.response.text()
        check_data['default'] = dict(json.loads(response))
        send_request_response = await context.send_request(str(server_url / 'user-agent'))
        check_data['send_request'] = dict(json.loads(send_request_response.read()))

    await crawler.run([str(server_url / 'user-agent')])

    assert check_data['default'].get('user-agent') is not None
    assert check_data['send_request']['user-agent'] == 'My User-Agent'

    assert check_data['default'] != check_data['send_request']


async def test_overwrite_configuration() -> None:
    """Check that the configuration is allowed to be passed to the Playwrightcrawler."""
    configuration = Configuration(default_dataset_id='my_dataset_id')
    PlaywrightCrawler(configuration=configuration)


Review comment (suggestion):
Please mention that the test only checks that the `PlaywrightCrawler` instantiation does not raise an error.

Code after:

async def test_send_request_with_client(server_url: URL) -> None:
    """Check that the persist context works with fingerprints."""
    check_data: dict[str, Any] = {}

    crawler = PlaywrightCrawler(
        http_client=HttpxHttpClient(header_generator=None, headers={'user-agent': 'My User-Agent'})
    )

    @crawler.router.default_handler
    async def request_handler(context: PlaywrightCrawlingContext) -> None:
        response = await context.response.text()
        check_data['default'] = dict(json.loads(response))
        send_request_response = await context.send_request(str(server_url / 'user-agent'))
        check_data['send_request'] = dict(json.loads(send_request_response.read()))

    await crawler.run([str(server_url / 'user-agent')])

    assert check_data['default'].get('user-agent') is not None
    assert check_data['send_request']['user-agent'] == 'My User-Agent'

    assert check_data['default'] != check_data['send_request']


async def test_overwrite_configuration() -> None:
    """Check that the configuration is allowed to be passed to the Playwrightcrawler."""
    configuration = Configuration(default_dataset_id='my_dataset_id')
    PlaywrightCrawler(configuration=configuration)
    used_configuration = service_locator.get_configuration()
    assert used_configuration is configuration
Repository: apify/crawlee-python
File: tests/unit/crawlers/_playwright/test_playwright_crawler.py (Python)

Code before:
    """Check that the persist context works with fingerprints."""
    check_data: dict[str, Any] = {}

    crawler = PlaywrightCrawler(
        http_client=HttpxHttpClient(header_generator=None, headers={'user-agent': 'My User-Agent'})
    )

    @crawler.router.default_handler
    async def request_handler(context: PlaywrightCrawlingContext) -> None:
        response = await context.response.text()
        check_data['default'] = dict(json.loads(response))
        send_request_response = await context.send_request(str(server_url / 'user-agent'))
        check_data['send_request'] = dict(json.loads(send_request_response.read()))

    await crawler.run([str(server_url / 'user-agent')])

    assert check_data['default'].get('user-agent') is not None
    assert check_data['send_request']['user-agent'] == 'My User-Agent'

    assert check_data['default'] != check_data['send_request']


async def test_overwrite_configuration() -> None:
    """Check that the configuration is allowed to be passed to the Playwrightcrawler."""
    configuration = Configuration(default_dataset_id='my_dataset_id')
    PlaywrightCrawler(configuration=configuration)


Review comment (suggestion):
Could you please add one assert that checks that such configuration is actually set when passed to the `PlaywrightCrawler`.

Code after:
    """Check that the persist context works with fingerprints."""
    check_data: dict[str, Any] = {}

    crawler = PlaywrightCrawler(
        http_client=HttpxHttpClient(header_generator=None, headers={'user-agent': 'My User-Agent'})
    )

    @crawler.router.default_handler
    async def request_handler(context: PlaywrightCrawlingContext) -> None:
        response = await context.response.text()
        check_data['default'] = dict(json.loads(response))
        send_request_response = await context.send_request(str(server_url / 'user-agent'))
        check_data['send_request'] = dict(json.loads(send_request_response.read()))

    await crawler.run([str(server_url / 'user-agent')])

    assert check_data['default'].get('user-agent') is not None
    assert check_data['send_request']['user-agent'] == 'My User-Agent'

    assert check_data['default'] != check_data['send_request']


async def test_overwrite_configuration() -> None:
    """Check that the configuration is allowed to be passed to the Playwrightcrawler."""
    configuration = Configuration(default_dataset_id='my_dataset_id')
    PlaywrightCrawler(configuration=configuration)
    used_configuration = service_locator.get_configuration()
    assert used_configuration is configuration
Repository: apify/crawlee-python
File: src/crawlee/project_template/{{cookiecutter.project_name}}/{{cookiecutter.__package_name}}/__main__.py (Python)

Code before:
import asyncio
# % if cookiecutter.http_client == 'curl-impersonate' and 'playwright' not in cookiecutter.crawler_type
import platform
# % endif
{{ '' }}
from .main import main

if __name__ == '__main__':
    # % if cookiecutter.http_client == 'curl-impersonate' and 'playwright' not in cookiecutter.crawler_type
    if platform.system() == 'Windows':
        # This mitigates a warning raised by curl-cffi.
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    # % endif
{{ '' }}
    asyncio.run(main())


Review comment (suggestion):
Maybe we should print a warning in case playwright+curl-cffi is selected, something like "Hey, curl-cffi will complain about the event loop policy, but that won't work with playwright, so just ignore the warning"

Code after:
import asyncio
# % if cookiecutter.http_client == 'curl-impersonate'
import platform
# % if 'playwright' in cookiecutter.crawler_type
import warnings
# % endif
# % endif
{{ '' }}
from .main import main

if __name__ == '__main__':
    # % if cookiecutter.http_client == 'curl-impersonate'
    if platform.system() == 'Windows':
        # This mitigates a warning raised by curl-cffi.
        # % if 'playwright' in cookiecutter.crawler_type
        warnings.warn(
            message=('curl-cffi suggests using WindowsSelectorEventLoopPolicy, but this conflicts with Playwright. '
                     'Ignore the curl-cffi warning.'),
            category=UserWarning,
            stacklevel=2,
        )
        # % else
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
        # % endif
    # % endif
{{ '' }}
    asyncio.run(main())
Repository: apify/crawlee-python
File: tests/unit/_statistics/test_request_processing_record.py (Python)

Code before:
from datetime import timedelta

from crawlee.statistics._statistics import RequestProcessingRecord


def test_tracking_time_resolution() -> None:
    """Test that `RequestProcessingRecord` tracks time with sufficient resolution

    This is generally not an issue on Linux, but on Windows some packages in older Python versions might be using system
    timers with not so granular resolution - some sources estimate 15ms. This test will start failing on Windows
    if unsuitable source of time measurement is selected due to two successive time measurements possibly using same
    timing sample."""
    record = RequestProcessingRecord()
    record.run()
    record.finish()
    assert record.duration
    assert record.duration > timedelta(seconds=0)


Review comment (suggestion):
```suggestion
    """Test that `RequestProcessingRecord` tracks time with sufficient resolution.
```

Code after:
from datetime import timedelta

from crawlee.statistics._statistics import RequestProcessingRecord


def test_tracking_time_resolution() -> None:
    """Test that `RequestProcessingRecord` tracks time with sufficient resolution.

    This is generally not an issue on Linux, but on Windows some packages in older Python versions might be using system
    timers with not so granular resolution - some sources estimate 15ms. This test will start failing on Windows
    if unsuitable source of time measurement is selected due to two successive time measurements possibly using same
    timing sample."""
    record = RequestProcessingRecord()
    record.run()
    record.finish()
    assert record.duration
    assert record.duration > timedelta(seconds=0)
Repository: apify/crawlee-python
File: src/crawlee/events/_types.py (Python)

Code before:
    # The remaining time in seconds before the migration is forced and the process is killed
    # Optional because it's not present when the event handler is called manually
    time_remaining: Annotated[timedelta_secs | None, Field(alias='timeRemainingSecs')] = None


@docs_group('Event payloads')
class EventAbortingData(BaseModel):
    """Data for the aborting event."""

    model_config = ConfigDict(populate_by_name=True)


@docs_group('Event payloads')
class EventExitData(BaseModel):
    """Data for the exit event."""

    model_config = ConfigDict(populate_by_name=True)


@docs_group('Event payloads')
class EventCrawlerStatusData(BaseModel):
    """Data for the crawler status event."""

    model_config = ConfigDict(populate_by_name=True)

    message: str


EventData = Union[
    EventPersistStateData,
    EventSystemInfoData,
    EventMigratingData,
    EventAbortingData,
    EventExitData,
    EventCrawlerStatusData,
]
"""A helper type for all possible event payloads"""

WrappedListener = Callable[..., Coroutine[Any, Any, None]]

TEvent = TypeVar('TEvent')
EventListener = Union[
    Callable[
        [TEvent],
        Union[None, Coroutine[Any, Any, None]],
    ],
    Callable[
        [],
        Union[None, Coroutine[Any, Any, None]],
    ],
]

Review comment (suggestion):
May I suggest adding something to distinguish crawlers for cases where there are multiple instances in a single run? If there's nothing more meaningful, `id(crawler)` should be fine.

Code after:
    # The remaining time in seconds before the migration is forced and the process is killed
    # Optional because it's not present when the event handler is called manually
    time_remaining: Annotated[timedelta_secs | None, Field(alias='timeRemainingSecs')] = None


@docs_group('Event payloads')
class EventAbortingData(BaseModel):
    """Data for the aborting event."""

    model_config = ConfigDict(populate_by_name=True)


@docs_group('Event payloads')
class EventExitData(BaseModel):
    """Data for the exit event."""

    model_config = ConfigDict(populate_by_name=True)


@docs_group('Event payloads')
class EventCrawlerStatusData(BaseModel):
    """Data for the crawler status event."""

    model_config = ConfigDict(populate_by_name=True)

    message: str
    """A message describing the current status of the crawler."""

    crawler_id: int
    """The ID of the crawler that emitted the event."""


EventData = (
    EventPersistStateData
    | EventSystemInfoData
    | EventMigratingData
    | EventAbortingData
    | EventExitData
    | EventCrawlerStatusData
)
"""A helper type for all possible event payloads"""

WrappedListener = Callable[..., Coroutine[Any, Any, None]]

TEvent = TypeVar('TEvent')
EventListener = (
    Callable[
        [TEvent],
        None | Coroutine[Any, Any, None],
    ]
    | Callable[
Repository: apify/crawlee-python
File: docs/guides/code_examples/playwright_crawler_stagehand/browser_classes.py (Python)

Code before:
        # Update internal state
        self._pages.append(pw_page)
        self._last_page_opened_at = datetime.now(timezone.utc)

        self._total_opened_pages += 1

        # Wrap StagehandPage to provide Playwright Page interface
        return cast('Page', CrawleeStagehandPage(page))


class StagehandPlugin(PlaywrightBrowserPlugin):
    """Browser plugin that integrates Stagehand with Crawlee's browser management."""

    @override
    def __init__(self, stagehand: Stagehand, **kwargs: Any) -> None:
        super().__init__(**kwargs)

        self._stagehand = stagehand

    @override
    async def new_browser(self) -> StagehandBrowserController:
        if not self._playwright:
            raise RuntimeError('Playwright browser plugin is not initialized.')

        browser = PlaywrightPersistentBrowser(
            self._playwright.chromium, self._user_data_dir, self._browser_launch_options
        )

        # Return custom controller with Stagehand
        return StagehandBrowserController(
            browser=browser,
            stagehand=self._stagehand,
            header_generator=None,
            fingerprint_generator=None,
        )


Review comment (suggestion):
I read this https://docs.stagehand.dev/examples/customize_browser

Based on the docs, it seems that, for instance, Firefox won't work here. Maybe it would be worth adding a comment like "Stagehand can run only on a Chromium-based browser." to make it clear that selected `chromium` was not just a random pick for the sake of example code.

Code after:
        # Update internal state
        self._pages.append(pw_page)
        self._last_page_opened_at = datetime.now(timezone.utc)

        self._total_opened_pages += 1

        # Wrap StagehandPage to provide Playwright Page interface
        return cast('Page', CrawleeStagehandPage(page))


class StagehandPlugin(PlaywrightBrowserPlugin):
    """Browser plugin that integrates Stagehand with Crawlee's browser management."""

    @override
    def __init__(self, stagehand: Stagehand, **kwargs: Any) -> None:
        super().__init__(**kwargs)

        self._stagehand = stagehand

    @override
    async def new_browser(self) -> StagehandBrowserController:
        if not self._playwright:
            raise RuntimeError('Playwright browser plugin is not initialized.')

        browser = PlaywrightPersistentBrowser(
            # Stagehand can run only on a Chromium-based browser.
            self._playwright.chromium,
            self._user_data_dir,
            self._browser_launch_options,
        )

        # Return custom controller with Stagehand
        return StagehandBrowserController(
            browser=browser,
            stagehand=self._stagehand,
            header_generator=None,
            fingerprint_generator=None,
        )
Repository: apify/crawlee-python
File: docs/guides/code_examples/playwright_crawler_stagehand/stagehand_run.py (Python)

Code before:
    crawler = PlaywrightCrawler(
        # Limit the crawl to max requests. Remove or increase it for crawling all links.
        max_requests_per_crawl=10,
        # Custom browser pool. Gives users full control over browsers used by the crawler.
        concurrency_settings=ConcurrencySettings(max_tasks_per_minute=10),
        browser_pool=BrowserPool(
            plugins=[
                StagehandPlugin(stagehand, browser_launch_options={'headless': True})
            ],
        ),
    )

    # Define the default request handler, which will be called for every request.
    @crawler.router.default_handler
    async def request_handler(context: PlaywrightCrawlingContext) -> None:
        context.log.info(f'Processing {context.request.url} ...')

        # Cast to StagehandPage for proper type hints in IDE
        page = cast('StagehandPage', context.page)

        # Use regular Playwright method
        playwright_title = await page.title()
        context.log.info(f'Playwright page title: {playwright_title}')

        # Use AI-powered extraction with natural language
        gemini_title = await page.extract('Extract title page')
        context.log.info(f'Gemini page title: {gemini_title}')

        await context.enqueue_links()

    # Run the crawler with the initial list of URLs.
    await crawler.run(['https://crawlee.dev/'])


if __name__ == '__main__':
    asyncio.run(main())


Review comment (refactor):
```suggestion
        gemini_title = await page.extract('Extract page title')
```

Code after:
    crawler = PlaywrightCrawler(
        # Limit the crawl to max requests. Remove or increase it for crawling all links.
        max_requests_per_crawl=10,
        # Custom browser pool. Gives users full control over browsers used by the crawler.
        concurrency_settings=ConcurrencySettings(max_tasks_per_minute=10),
        browser_pool=BrowserPool(
            plugins=[
                StagehandPlugin(stagehand, browser_launch_options={'headless': True})
            ],
        ),
    )

    # Define the default request handler, which will be called for every request.
    @crawler.router.default_handler
    async def request_handler(context: PlaywrightCrawlingContext) -> None:
        context.log.info(f'Processing {context.request.url} ...')

        # Cast to StagehandPage for proper type hints in IDE
        page = cast('StagehandPage', context.page)

        # Use regular Playwright method
        playwright_title = await page.title()
        context.log.info(f'Playwright page title: {playwright_title}')

        # highlight-start
        # Use AI-powered extraction with natural language
        gemini_title = await page.extract('Extract page title')
        context.log.info(f'Gemini page title: {gemini_title}')
        # highlight-end

        await context.enqueue_links()

    # Run the crawler with the initial list of URLs.
    await crawler.run(['https://crawlee.dev/'])


if __name__ == '__main__':
    asyncio.run(main())
Repository: apify/crawlee-python
File: docs/guides/code_examples/playwright_crawler_stagehand/stagehand_run.py (Python)

Code before:
        max_requests_per_crawl=10,
        # Custom browser pool. Gives users full control over browsers used by the crawler.
        concurrency_settings=ConcurrencySettings(max_tasks_per_minute=10),
        browser_pool=BrowserPool(
            plugins=[
                StagehandPlugin(stagehand, browser_launch_options={'headless': True})
            ],
        ),
    )

    # Define the default request handler, which will be called for every request.
    @crawler.router.default_handler
    async def request_handler(context: PlaywrightCrawlingContext) -> None:
        context.log.info(f'Processing {context.request.url} ...')

        # Cast to StagehandPage for proper type hints in IDE
        page = cast('StagehandPage', context.page)

        # Use regular Playwright method
        playwright_title = await page.title()
        context.log.info(f'Playwright page title: {playwright_title}')

        # Use AI-powered extraction with natural language
        gemini_title = await page.extract('Extract page title')
        context.log.info(f'Gemini page title: {gemini_title}')

        await context.enqueue_links()

    # Run the crawler with the initial list of URLs.
    await crawler.run(['https://crawlee.dev/'])


if __name__ == '__main__':
    asyncio.run(main())


Review comment (refactor):
Could we highlight this?

```suggestion
        # highlight-start
        # Use AI-powered extraction with natural language
        gemini_title = await page.extract('Extract page title')
        context.log.info(f'Gemini page title: {gemini_title}')
        # highlight-end
```

Code after:
        max_requests_per_crawl=10,
        # Custom browser pool. Gives users full control over browsers used by the crawler.
        concurrency_settings=ConcurrencySettings(max_tasks_per_minute=10),
        browser_pool=BrowserPool(
            plugins=[
                StagehandPlugin(stagehand, browser_launch_options={'headless': True})
            ],
        ),
    )

    # Define the default request handler, which will be called for every request.
    @crawler.router.default_handler
    async def request_handler(context: PlaywrightCrawlingContext) -> None:
        context.log.info(f'Processing {context.request.url} ...')

        # Cast to StagehandPage for proper type hints in IDE
        page = cast('StagehandPage', context.page)

        # Use regular Playwright method
        playwright_title = await page.title()
        context.log.info(f'Playwright page title: {playwright_title}')

        # highlight-start
        # Use AI-powered extraction with natural language
        gemini_title = await page.extract('Extract page title')
        context.log.info(f'Gemini page title: {gemini_title}')
        # highlight-end

        await context.enqueue_links()

    # Run the crawler with the initial list of URLs.
    await crawler.run(['https://crawlee.dev/'])


if __name__ == '__main__':
    asyncio.run(main())