ronantakizawa/codereview-bench huggingface dataset:
Repository: apify/crawlee-python
File: src/crawlee/_utils/raise_if_too_many_kwargs.py (Python)

Code before:
from collections import Counter
from typing import Any


def raise_if_too_many_kwargs(max_kwargs: int = 1, **kwargs: Any) -> None:
    """Raise ValueError if there are more non-None kwargs then max_kwargs."""
    if len(kwargs) - Counter(kwargs.values())[None] > max_kwargs:
        kwargs_names = [f'"{kwarg_name}"' for kwarg_name in kwargs]
        used_kwargs_names = [f'"{kwarg_name}"' for kwarg_name, value in kwargs.items() if value is not None]
        raise ValueError(
            f'Only one of {", ".join(kwargs_names)} can be specified, but following arguments were '
            f'specified: {", ".join(used_kwargs_names)}.'
        )


Review comment (suggestion):
Since we need to use list comprehension for `used_kwargs_names` anyway, do we really need Counter?

maybe just
```python
used_kwargs_names = [f'"{kwarg_name}"' for kwarg_name, value in kwargs.items() if value is not None]
if len(used_kwargs_names ) > max_kwargs:
```

Code after:
from typing import Any


def raise_if_too_many_kwargs(max_kwargs: int = 1, **kwargs: Any) -> None:
    """Raise ValueError if there are more non-None kwargs then max_kwargs."""
    none_kwargs_names = [f'"{kwarg_name}"' for kwarg_name, value in kwargs.items() if value is not None]
    if len(none_kwargs_names) > max_kwargs:
        all_kwargs_names = [f'"{kwarg_name}"' for kwarg_name in kwargs]
        raise ValueError(
            f'Only one of {", ".join(all_kwargs_names)} can be specified, but following arguments were '
            f'specified: {", ".join(none_kwargs_names)}.'
        )
Repository: apify/crawlee-python
File: tests/unit/browsers/test_playwright_browser_plugin.py (Python)

Code before:

    assert browser_controller_1 is not browser_controller_2


async def test_methods_raise_error_when_not_active() -> None:
    plugin = PlaywrightBrowserPlugin()

    assert plugin.active is False

    with pytest.raises(RuntimeError, match=r'Plugin is not active'):
        await plugin.new_browser()

    with pytest.raises(RuntimeError, match=r'Plugin is already active.'):
        async with plugin, plugin:
            pass

    async with plugin:
        assert plugin.active is True


async def raise_error_if_use_chrome_and_executable_path() -> None:
    with pytest.raises(
        ValueError, match=r'Cannot use `use_chrome` with `Configuration.default_browser_path` or `executable_path` set.'
    ):
        PlaywrightBrowserPlugin(
            browser_type='chromium',
            use_chrome=True,
            browser_launch_options={'executable_path': '/path/to/chrome'},
        )


async def raise_error_if_use_chrome_with_firefox() -> None:
    with pytest.raises(ValueError, match=r'`use_chrome` can only be used with `chromium` `browser_type`.'):
        PlaywrightBrowserPlugin(
            browser_type='firefox',
            use_chrome=True,
        )


Review comment (suggestion):
To me, this seems a little strange from the user perspective. It is defining one type of browser in one argument and then overriding it in another argument, which is valid only for a specific value of the first argument. I would prefer to just add browser_type=`chrome` and connect any logic to such a specific browser_type internally.

Code after:

    assert browser_controller_1 is not browser_controller_2


async def test_methods_raise_error_when_not_active() -> None:
    plugin = PlaywrightBrowserPlugin()

    assert plugin.active is False

    with pytest.raises(RuntimeError, match=r'Plugin is not active'):
        await plugin.new_browser()

    with pytest.raises(RuntimeError, match=r'Plugin is already active.'):
        async with plugin, plugin:
            pass

    async with plugin:
        assert plugin.active is True


async def raise_error_if_chrome_and_executable_path() -> None:
    with pytest.raises(
        ValueError, match=r'Cannot use `use_chrome` with `Configuration.default_browser_path` or `executable_path` set.'
    ):
        PlaywrightBrowserPlugin(
            browser_type='chrome',
            browser_launch_options={'executable_path': '/path/to/chrome'},
        )
Repository: apify/crawlee-python
File: src/crawlee/crawlers/_adaptive_playwright/_adaptive_playwright_crawler.py (Python)

Code before:
    def track_rendering_type_mispredictions(self) -> None:
        self.statistics.state.rendering_type_mispredictions += 1

    def _update_context_from_copy(self, context: BasicCrawlingContext, context_copy: BasicCrawlingContext) -> None:
        """Update mutable fields of `context` from `context_copy`.

        Uses object.__setattr__ to bypass frozen dataclass restrictions,
        allowing state synchronization after isolated crawler execution.
        """
        updating_attributes = ('request', 'session', 'proxy_info')

        for attr_name in updating_attributes:
            original_obj = getattr(context, attr_name, None)
            updated_obj = getattr(context_copy, attr_name, None)

            # Skip if either object is None or they are the same object
            if (
                original_obj is None
                or updated_obj is None
                or original_obj is updated_obj
                or not hasattr(updated_obj, '__dict__')
            ):
                continue

            # Copy all attributes using __dict__
            for obj_attr_name, obj_attr_value in updated_obj.__dict__.items():
                object.__setattr__(original_obj, obj_attr_name, obj_attr_value)


@dataclass(frozen=True)
class SubCrawlerRun:
    result: RequestHandlerRunResult | None = None
    exception: Exception | None = None
    run_context: BasicCrawlingContext | None = None


Review comment (suggestion):
This is quite an overkill. I'd much rather just enumerate the attributes for which it makes sense to mutate them from the request handler.

Code after:
        """Update mutable fields of `context` from `context_copy`.

        Uses object.__setattr__ to bypass frozen dataclass restrictions,
        allowing state synchronization after isolated crawler execution.
        """
        updating_attributes = {
            'request': ('headers', 'user_data'),
            'session': ('_user_data', '_usage_count', '_error_score', '_cookies'),
        }

        for attr, sub_attrs in updating_attributes.items():
            original_sub_obj = getattr(context, attr)
            copy_sub_obj = getattr(context_copy, attr)

            # Check that both sub objects are not None
            if original_sub_obj is None or copy_sub_obj is None:
                continue

            for sub_attr in sub_attrs:
                new_value = getattr(copy_sub_obj, sub_attr)
                object.__setattr__(original_sub_obj, sub_attr, new_value)


@dataclass(frozen=True)
class SubCrawlerRun:
    result: RequestHandlerRunResult | None = None
    exception: Exception | None = None
    run_context: BasicCrawlingContext | None = None
Repository: apify/crawlee-python
File: tests/unit/request_loaders/test_sitemap_request_loader.py (Python)

Code before:
    next_item_in_kvs = state_data['urlQueue'][0]

    sitemap_loader = SitemapRequestLoader([str(sitemap_url)], http_client=http_client, persist_state_key=persist_key)

    item = await sitemap_loader.fetch_next_request()

    assert item is not None
    assert item.url == next_item_in_kvs


async def test_transform_request_function(server_url: URL, http_client: HttpClient) -> None:
    sitemap_url = (server_url / 'sitemap.xml').with_query(base64=encode_base64(BASIC_SITEMAP.encode()))

    def transform_request(request_options: RequestOptions) -> RequestOptions | RequestTransformAction:
        request_options['user_data'] = {'transformed': True}
        return request_options

    sitemap_loader = SitemapRequestLoader(
        [str(sitemap_url)],
        http_client=http_client,
        transform_request_function=transform_request,
    )

    while not await sitemap_loader.is_finished():
        request = await sitemap_loader.fetch_next_request()
        assert request is not None
        assert request.user_data.get('transformed') is True

        await sitemap_loader.mark_request_as_handled(request)


Review comment (refactor):
Can we add an assertion about the correct number of requests extracted from the BASIC_SITEMAP?

if sitemap_loader.is_finished() returns True and no request was extracted, then this test still passes

Code after:
    next_item_in_kvs = state_data['urlQueue'][0]

    sitemap_loader = SitemapRequestLoader([str(sitemap_url)], http_client=http_client, persist_state_key=persist_key)

    item = await sitemap_loader.fetch_next_request()

    assert item is not None
    assert item.url == next_item_in_kvs


async def test_transform_request_function(server_url: URL, http_client: HttpClient) -> None:
    sitemap_url = (server_url / 'sitemap.xml').with_query(base64=encode_base64(BASIC_SITEMAP.encode()))

    def transform_request(request_options: RequestOptions) -> RequestOptions | RequestTransformAction:
        request_options['user_data'] = {'transformed': True}
        return request_options

    sitemap_loader = SitemapRequestLoader(
        [str(sitemap_url)],
        http_client=http_client,
        transform_request_function=transform_request,
    )

    extracted_urls = set()

    while not await sitemap_loader.is_finished():
        request = await sitemap_loader.fetch_next_request()
        assert request is not None
        assert request.user_data.get('transformed') is True

        extracted_urls.add(request.url)

        await sitemap_loader.mark_request_as_handled(request)

    assert len(extracted_urls) == 5
    assert extracted_urls == {
        'http://not-exists.com/',
        'http://not-exists.com/catalog?item=12&desc=vacation_hawaii',
        'http://not-exists.com/catalog?item=73&desc=vacation_new_zealand',
        'http://not-exists.com/catalog?item=74&desc=vacation_newfoundland',
        'http://not-exists.com/catalog?item=83&desc=vacation_usa',
    }
Repository: apify/crawlee-python
File: tests/unit/crawlers/_basic/test_basic_crawler.py (Python)

Code before:

    request = Request.from_url('https://a.placeholder.com')

    @crawler.router.default_handler
    async def handler(context: BasicCrawlingContext) -> None:
        if '|test' in context.request.unique_key:
            return
        raise ValueError('This error should not be handled by error handler')

    @crawler.error_handler
    async def error_handler(context: BasicCrawlingContext, error: Exception) -> Request | None:
        return Request.from_url(
            context.request.url,
            unique_key=f'{context.request.unique_key}|test',
        )

    await crawler.run([request])

    check_original_request = await queue.get_request(request.unique_key)
    check_error_request = await queue.get_request(f'{request.unique_key}|test')

    assert check_original_request is not None
    assert check_original_request.state == RequestState.ERROR_HANDLER

    assert check_error_request is not None
    assert check_error_request.state == RequestState.REQUEST_HANDLER


Review comment (suggestion):
```suggestion
    original_request = await queue.get_request(request.unique_key)
    error_request = await queue.get_request(f'{request.unique_key}|test')

    assert original_request is not None
    assert original_request.state == RequestState.ERROR_HANDLER
    assert original_request.was_already_handled

    assert error_request is not None
    assert error_request.state == RequestState.REQUEST_HANDLER
    assert error_request.was_already_handled
```

Code after:

    request = Request.from_url('https://a.placeholder.com')

    @crawler.router.default_handler
    async def handler(context: BasicCrawlingContext) -> None:
        if '|test' in context.request.unique_key:
            return
        raise ValueError('This error should not be handled by error handler')

    @crawler.error_handler
    async def error_handler(context: BasicCrawlingContext, error: Exception) -> Request | None:
        return Request.from_url(
            context.request.url,
            unique_key=f'{context.request.unique_key}|test',
        )

    await crawler.run([request])

    original_request = await queue.get_request(request.unique_key)
    error_request = await queue.get_request(f'{request.unique_key}|test')

    assert original_request is not None
    assert original_request.state == RequestState.ERROR_HANDLER
    assert original_request.was_already_handled

    assert error_request is not None
    assert error_request.state == RequestState.REQUEST_HANDLER
    assert error_request.was_already_handled
Repository: apify/crawlee-python
File: tests/unit/crawlers/_beautifulsoup/test_beautifulsoup_crawler.py (Python)

Code before:
    assert result.requests_finished == 0
    assert request_handler.call_count == 0


async def test_slow_navigation_does_not_count_toward_handler_timeout(server_url: URL, http_client: HttpClient) -> None:
    crawler = BeautifulSoupCrawler(
        http_client=http_client,
        request_handler_timeout=timedelta(seconds=0.5),
        max_request_retries=0,
    )

    request_handler = mock.AsyncMock()
    crawler.router.default_handler(request_handler)

    # Navigation takes 1 second (exceeds handler timeout), but should still succeed
    result = await crawler.run([str(server_url.with_path('/slow').with_query(delay=1))])

    assert result.requests_failed == 0
    assert result.requests_finished == 1
    assert request_handler.call_count == 1


async def test_enqueue_strategy_after_redirect(server_url: URL, redirect_server_url: URL) -> None:
    crawler = BeautifulSoupCrawler()

    hendler_calls = mock.AsyncMock()

    @crawler.router.default_handler
    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
        await hendler_calls(context.request.url)

        target_url = str(server_url.with_path('redirect').with_query(url=str(redirect_server_url)))

        await context.enqueue_links(requests=[Request.from_url(target_url)], strategy='same-origin')

    await crawler.run([str(server_url)])

    assert hendler_calls.called
    assert hendler_calls.call_count == 1


Review comment (suggestion):
```suggestion
    handles_calls = mock.AsyncMock()
```

Code after:
    assert result.requests_finished == 0
    assert request_handler.call_count == 0


async def test_slow_navigation_does_not_count_toward_handler_timeout(server_url: URL, http_client: HttpClient) -> None:
    crawler = BeautifulSoupCrawler(
        http_client=http_client,
        request_handler_timeout=timedelta(seconds=0.5),
        max_request_retries=0,
    )

    request_handler = mock.AsyncMock()
    crawler.router.default_handler(request_handler)

    # Navigation takes 1 second (exceeds handler timeout), but should still succeed
    result = await crawler.run([str(server_url.with_path('/slow').with_query(delay=1))])

    assert result.requests_failed == 0
    assert result.requests_finished == 1
    assert request_handler.call_count == 1


async def test_enqueue_strategy_after_redirect(server_url: URL, redirect_server_url: URL) -> None:
    crawler = BeautifulSoupCrawler()

    handler_calls = mock.AsyncMock()

    @crawler.router.default_handler
    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
        await handler_calls(context.request.url)

        target_url = str(server_url.with_path('redirect').with_query(url=str(redirect_server_url)))

        await context.enqueue_links(requests=[Request.from_url(target_url)], strategy='same-origin')

    await crawler.run([str(server_url)])

    assert handler_calls.called
    assert handler_calls.call_count == 1
Repository: apify/crawlee-python
File: tests/unit/crawlers/_beautifulsoup/test_beautifulsoup_crawler.py (Python)

Code before:
async def test_slow_navigation_does_not_count_toward_handler_timeout(server_url: URL, http_client: HttpClient) -> None:
    crawler = BeautifulSoupCrawler(
        http_client=http_client,
        request_handler_timeout=timedelta(seconds=0.5),
        max_request_retries=0,
    )

    request_handler = mock.AsyncMock()
    crawler.router.default_handler(request_handler)

    # Navigation takes 1 second (exceeds handler timeout), but should still succeed
    result = await crawler.run([str(server_url.with_path('/slow').with_query(delay=1))])

    assert result.requests_failed == 0
    assert result.requests_finished == 1
    assert request_handler.call_count == 1


async def test_enqueue_strategy_after_redirect(server_url: URL, redirect_server_url: URL) -> None:
    crawler = BeautifulSoupCrawler()

    hendler_calls = mock.AsyncMock()

    @crawler.router.default_handler
    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
        await hendler_calls(context.request.url)

        target_url = str(server_url.with_path('redirect').with_query(url=str(redirect_server_url)))

        await context.enqueue_links(requests=[Request.from_url(target_url)], strategy='same-origin')

    await crawler.run([str(server_url)])

    assert hendler_calls.called
    assert hendler_calls.call_count == 1


Review comment (suggestion):
```suggestion
        await handler_calls(context.request.url)
```

Code after:
async def test_slow_navigation_does_not_count_toward_handler_timeout(server_url: URL, http_client: HttpClient) -> None:
    crawler = BeautifulSoupCrawler(
        http_client=http_client,
        request_handler_timeout=timedelta(seconds=0.5),
        max_request_retries=0,
    )

    request_handler = mock.AsyncMock()
    crawler.router.default_handler(request_handler)

    # Navigation takes 1 second (exceeds handler timeout), but should still succeed
    result = await crawler.run([str(server_url.with_path('/slow').with_query(delay=1))])

    assert result.requests_failed == 0
    assert result.requests_finished == 1
    assert request_handler.call_count == 1


async def test_enqueue_strategy_after_redirect(server_url: URL, redirect_server_url: URL) -> None:
    crawler = BeautifulSoupCrawler()

    handler_calls = mock.AsyncMock()

    @crawler.router.default_handler
    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
        await handler_calls(context.request.url)

        target_url = str(server_url.with_path('redirect').with_query(url=str(redirect_server_url)))

        await context.enqueue_links(requests=[Request.from_url(target_url)], strategy='same-origin')

    await crawler.run([str(server_url)])

    assert handler_calls.called
    assert handler_calls.call_count == 1
Repository: apify/crawlee-python
File: tests/unit/crawlers/_beautifulsoup/test_beautifulsoup_crawler.py (Python)

Code before:

    # Navigation takes 1 second (exceeds handler timeout), but should still succeed
    result = await crawler.run([str(server_url.with_path('/slow').with_query(delay=1))])

    assert result.requests_failed == 0
    assert result.requests_finished == 1
    assert request_handler.call_count == 1


async def test_enqueue_strategy_after_redirect(server_url: URL, redirect_server_url: URL) -> None:
    crawler = BeautifulSoupCrawler()

    hendler_calls = mock.AsyncMock()

    @crawler.router.default_handler
    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
        await hendler_calls(context.request.url)

        target_url = str(server_url.with_path('redirect').with_query(url=str(redirect_server_url)))

        await context.enqueue_links(requests=[Request.from_url(target_url)], strategy='same-origin')

    await crawler.run([str(server_url)])

    assert hendler_calls.called
    assert hendler_calls.call_count == 1


Review comment (suggestion):
```suggestion
    assert handler_calls.called
    assert handler_calls.call_count == 1
```

Code after:

    # Navigation takes 1 second (exceeds handler timeout), but should still succeed
    result = await crawler.run([str(server_url.with_path('/slow').with_query(delay=1))])

    assert result.requests_failed == 0
    assert result.requests_finished == 1
    assert request_handler.call_count == 1


async def test_enqueue_strategy_after_redirect(server_url: URL, redirect_server_url: URL) -> None:
    crawler = BeautifulSoupCrawler()

    handler_calls = mock.AsyncMock()

    @crawler.router.default_handler
    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:
        await handler_calls(context.request.url)

        target_url = str(server_url.with_path('redirect').with_query(url=str(redirect_server_url)))

        await context.enqueue_links(requests=[Request.from_url(target_url)], strategy='same-origin')

    await crawler.run([str(server_url)])

    assert handler_calls.called
    assert handler_calls.call_count == 1
Repository: apify/crawlee-python
File: src/crawlee/_utils/byte_size.py (Python)

Code before:
from __future__ import annotations

from dataclasses import dataclass
from typing import Annotated, Any

from pydantic import BaseModel, Field

_BYTES_PER_KB = 1024
_BYTES_PER_MB = _BYTES_PER_KB**2
_BYTES_PER_GB = _BYTES_PER_KB**3
_BYTES_PER_TB = _BYTES_PER_KB**4


class Ratio(BaseModel):
    """Represents ratio of memory."""

    value: Annotated[float, Field(gt=0.0, le=1.0)]


@dataclass(frozen=True)
class ByteSize:
    """Represents a byte size."""

    bytes: int

    def __post_init__(self) -> None:

Review comment (suggestion):
I don't think the `Ratio` belongs here. A ratio is not a byte size - it's a proportion of memory. We should place `Ration` in the `_autoscaling/_types.py` where it is (only) used.

Code after:
from __future__ import annotations

from dataclasses import dataclass
from typing import Any

_BYTES_PER_KB = 1024
_BYTES_PER_MB = _BYTES_PER_KB**2
_BYTES_PER_GB = _BYTES_PER_KB**3
_BYTES_PER_TB = _BYTES_PER_KB**4


@dataclass(frozen=True)
class ByteSize:
    """Represents a byte size."""

    bytes: int

    def __post_init__(self) -> None:
        if self.bytes < 0:
            raise ValueError('ByteSize cannot be negative')

    @classmethod
    def validate(cls, value: Any) -> ByteSize:
        if isinstance(value, ByteSize):
            return value
Repository: apify/crawlee-python
File: src/crawlee/project_template/templates/main_playwright_chrome.py (Python)

Code before:
# % extends 'main.py'

# % block import
from crawlee.crawlers import PlaywrightCrawler
# % endblock

# % block instantiation
crawler = PlaywrightCrawler(
    request_handler=router,
    headless=True,
    max_requests_per_crawl=10,
    browser_type="chrome",
    {{ self.http_client_instantiation() }})
# % endblock


Review comment (suggestion):
Nit: Closing parenthesis formatting

  Files: templates/main_playwright_chrome.py,
  templates/main_playwright_firefox.py, templates/main_playwright_webkit.py

  The closing ) is on the same line as the last argument:

      {{ self.http_client_instantiation() }})

  Consider:

      {{ self.http_client_instantiation() }}
  )

Code after:
# % extends 'main.py'

# % block import
from crawlee.crawlers import PlaywrightCrawler
# % endblock

# % block instantiation
crawler = PlaywrightCrawler(
    request_handler=router,
    headless=True,
    max_requests_per_crawl=10,
    browser_type="chrome",
    {{ self.http_client_instantiation() }}
)
# % endblock