ronantakizawa/codereview-bench huggingface dataset:
Repository: pydantic/pydantic-ai
File: tests/test_exceptions.py (Python)

Code before:
        'UserError',
        'AgentRunError',
        'UnexpectedModelBehavior',
        'UsageLimitExceeded',
        'ModelAPIError',
        'ModelHTTPError',
        'IncompleteToolCall',
        'ToolRetryError',
    ],
)
def test_exceptions_hashable(exc_factory: Callable[[], Any]):
    """Test that all exception classes are hashable and usable as keys."""
    exc = exc_factory()

    # Does not raise TypeError
    _ = hash(exc)

    # Can be used in sets and dicts
    s = {exc}
    d = {exc: 'value'}

    assert exc in s
    assert d[exc] == 'value'


def test_tool_retry_error_str():
    """Test that ToolRetryError uses provided message."""
    part = RetryPromptPart(content='some content', tool_name='my_tool')
    error = ToolRetryError(part, message='custom message')
    assert str(error) == 'custom message'


Review comment (suggestion):
We should test that it is set correctly in the `ValidationError` case

Code after:
        'CallDeferred',
        'ApprovalRequired',
        'UserError',
        'AgentRunError',
        'UnexpectedModelBehavior',
        'UsageLimitExceeded',
        'ModelAPIError',
        'ModelHTTPError',
        'IncompleteToolCall',
        'ToolRetryError',
    ],
)
def test_exceptions_hashable(exc_factory: Callable[[], Any]):
    """Test that all exception classes are hashable and usable as keys."""
    exc = exc_factory()

    # Does not raise TypeError
    _ = hash(exc)

    # Can be used in sets and dicts
    s = {exc}
    d = {exc: 'value'}

    assert exc in s
    assert d[exc] == 'value'


def test_tool_retry_error_str_with_string_content():
    """Test that ToolRetryError uses string content as message automatically."""
    part = RetryPromptPart(content='error from tool', tool_name='my_tool')
    error = ToolRetryError(part)
    assert str(error) == 'error from tool'


def test_tool_retry_error_str_with_error_details():
    """Test that ToolRetryError formats ErrorDetails automatically."""
    validation_error = ValidationError.from_exception_data(
        'Test', [{'type': 'string_type', 'loc': ('name',), 'input': 123}]
    )
    part = RetryPromptPart(content=validation_error.errors(include_url=False), tool_name='my_tool')
    error = ToolRetryError(part)

    assert str(error) == (
        "1 validation error for 'my_tool'\nname\n  Input should be a valid string [type=string_type, input_value=123]"
    )


def test_tool_retry_error_str_with_value_error_type():
    """Test that ToolRetryError handles value_error type without ctx.error.

    When ErrorDetails are serialized, the exception object in ctx is stripped.
Repository: pydantic/pydantic-ai
File: pydantic_ai_slim/pydantic_ai/builtin_tools.py (Python)

Code before:
    """

    partial_images: int = 0
    """
    Number of partial images to generate in streaming mode.

    Supported by:

    * OpenAI Responses. Supports 0 to 3.
    """

    quality: Literal['low', 'medium', 'high', 'auto'] = 'auto'
    """The quality of the generated image.

    Supported by:

    * OpenAI Responses
    """

    size: Literal['1024x1024', '1024x1536', '1536x1024', '1K', '2K', '4K', 'auto'] = 'auto'
    """The size of the generated image.

    Supported by:

    * OpenAI Responses: '1024x1024', '1024x1536', '1536x1024', 'auto'
    * Google (Gemini 3 Pro Image and later): '1K', '2K', '4K'
    """

    aspect_ratio: ImageAspectRatio | None = None
    """The aspect ratio to use for generated images.

    Supported by:

    * Google image-generation models (Gemini)
    * OpenAI Responses (maps '1:1', '2:3', and '3:2' to supported sizes)
    """

    kind: str = 'image_generation'
    """The kind of tool."""


@dataclass(kw_only=True)
class MemoryTool(AbstractBuiltinTool):
    """A builtin tool that allows your agent to use memory.

    Supported by:

    * Anthropic
    """

    kind: str = 'memory'

Review comment (suggestion):
We effectively support `auto` on Google as well by not sending any value to the API, right? So I'd drop `auto` from the OpenAI Responses list above as it's not really OpenAI specific

Code after:

    output_compression: int = 100
    """Compression level for the output image.

    Supported by:

    * OpenAI Responses. Only supported for 'png' and 'webp' output formats.
    """

    output_format: Literal['png', 'webp', 'jpeg'] | None = None
    """The output format of the generated image.

    Supported by:

    * OpenAI Responses. Default: 'png'.
    """

    partial_images: int = 0
    """
    Number of partial images to generate in streaming mode.

    Supported by:

    * OpenAI Responses. Supports 0 to 3.
    """

    quality: Literal['low', 'medium', 'high', 'auto'] = 'auto'
    """The quality of the generated image.

    Supported by:

    * OpenAI Responses
    """

    size: Literal['auto', '1024x1024', '1024x1536', '1536x1024', '1K', '2K', '4K'] | None = None
    """The size of the generated image.

    * OpenAI Responses: 'auto' (default: model selects the size based on the prompt), '1024x1024', '1024x1536', '1536x1024'
    * Google (Gemini 3 Pro Image and later): '1K' (default), '2K', '4K'
    """

    aspect_ratio: ImageAspectRatio | None = None
    """The aspect ratio to use for generated images.

    Supported by:

    * Google image-generation models (Gemini)
    * OpenAI Responses (maps '1:1', '2:3', and '3:2' to supported sizes)
    """

    kind: str = 'image_generation'
Repository: pydantic/pydantic-ai
File: pydantic_ai_slim/pydantic_ai/ui/vercel_ai/response_types.py (Python)

Code before:
    """If true, data is not persisted to the message history."""


class StartStepChunk(BaseChunk):
    """Start step chunk."""

    type: Literal['start-step'] = 'start-step'


class FinishStepChunk(BaseChunk):
    """Finish step chunk."""

    type: Literal['finish-step'] = 'finish-step'


class StartChunk(BaseChunk):
    """Start chunk."""

    type: Literal['start'] = 'start'
    message_id: str | None = None
    message_metadata: Any | None = None


class FinishChunk(BaseChunk):
    """Finish chunk."""

    type: Literal['finish'] = 'finish'
    finish_reason: FinishReason = None
    """Reason why the model finished generating."""
    message_metadata: Any | None = None


class AbortChunk(BaseChunk):
    """Abort chunk."""

    type: Literal['abort'] = 'abort'


class MessageMetadataChunk(BaseChunk):
    """Message metadata chunk."""

    type: Literal['message-metadata'] = 'message-metadata'
    message_metadata: Any


class DoneChunk(BaseChunk):
    """Done chunk."""

    type: Literal['done'] = 'done'

    def encode(self) -> str:

Review comment (suggestion):
Can you please update the `after_stream` method that emits `FinishChunk` to set this field based on `ModelResponse.finish_reason`? That method doesn't take the model response directly, but is guaranteed to happen _after_ the `AgentRunResultEvent` handler is called, so we can set an instance var there and read it from `after_stream`.

Code after:

class StartStepChunk(BaseChunk):
    """Start step chunk."""

    type: Literal['start-step'] = 'start-step'


class FinishStepChunk(BaseChunk):
    """Finish step chunk."""

    type: Literal['finish-step'] = 'finish-step'


class StartChunk(BaseChunk):
    """Start chunk."""

    type: Literal['start'] = 'start'
    message_id: str | None = None
    message_metadata: Any | None = None


class FinishChunk(BaseChunk):
    """Finish chunk."""

    type: Literal['finish'] = 'finish'
    finish_reason: FinishReason = None
    message_metadata: Any | None = None


class AbortChunk(BaseChunk):
    """Abort chunk."""

    type: Literal['abort'] = 'abort'


class MessageMetadataChunk(BaseChunk):
    """Message metadata chunk."""

    type: Literal['message-metadata'] = 'message-metadata'
    message_metadata: Any


class DoneChunk(BaseChunk):
    """Done chunk."""

    type: Literal['done'] = 'done'

    def encode(self) -> str:
        return '[DONE]'
Repository: pydantic/pydantic-ai
File: tests/models/test_openrouter.py (Python)

Code before:
import datetime
import json
from collections.abc import Sequence
from typing import TYPE_CHECKING, Literal, cast

import pytest
from inline_snapshot import snapshot
from pydantic import BaseModel

if TYPE_CHECKING:
    from vcr.cassette import Cassette

from pydantic_ai import (
    Agent,
    BinaryContent,
    DocumentUrl,
    ModelHTTPError,
    ModelMessage,
    ModelRequest,
    ModelResponse,
    PartEndEvent,
    PartStartEvent,
    RunUsage,
    TextPart,
    ThinkingPart,
    ToolCallPart,
    ToolDefinition,
    UnexpectedModelBehavior,
)
from pydantic_ai.direct import model_request, model_request_stream
from pydantic_ai.models import ModelRequestParameters

from ..conftest import try_import

with try_import() as imports_successful:
    from openai.types.chat import ChatCompletion
    from openai.types.chat.chat_completion import Choice

Review comment (suggestion):
do you have something against this? I'm generally for marking type-hint-only classes explicitly like this

Code after:
import datetime
import json
from collections.abc import Sequence
from typing import Literal, cast

import pytest
from inline_snapshot import snapshot
from pydantic import BaseModel
from vcr.cassette import Cassette

from pydantic_ai import (
    Agent,
    BinaryContent,
    DocumentUrl,
    ModelHTTPError,
    ModelMessage,
    ModelRequest,
    ModelResponse,
    PartEndEvent,
    PartStartEvent,
    RunUsage,
    TextPart,
    ThinkingPart,
    ToolCallPart,
    ToolDefinition,
    UnexpectedModelBehavior,
)
from pydantic_ai.direct import model_request, model_request_stream
from pydantic_ai.models import ModelRequestParameters

from ..conftest import try_import

with try_import() as imports_successful:
    from openai.types.chat import ChatCompletion
    from openai.types.chat.chat_completion import Choice

    from pydantic_ai.models.openrouter import OpenRouterModel, OpenRouterModelSettings
Repository: pydantic/pydantic-ai
File: tests/models/test_google.py (Python)

Code before:

    captured_timeout: float | None = None

    mock_response = mocker.MagicMock()
    mock_response.status_code = 200
    mock_response.headers = {}
    mock_response.text = json.dumps(
        {
            'candidates': [{'content': {'parts': [{'text': 'Hello!'}], 'role': 'model'}, 'finishReason': 'STOP'}],
            'usageMetadata': {'promptTokenCount': 1, 'candidatesTokenCount': 1, 'totalTokenCount': 2},
        }
    )

    async def capture_timeout_request(*args: Any, **kwargs: Any):
        nonlocal captured_timeout
        captured_timeout = kwargs.get('timeout')
        return mock_response

    httpx_client = provider._client._api_client._async_httpx_client  # pyright: ignore[reportPrivateUsage]
    mocker.patch.object(
        httpx_client,
        'request',
        side_effect=capture_timeout_request,
    )

    await agent.run('Hello!')

    assert captured_timeout is not None, 'Timeout was None, which would cause requests to hang indefinitely'
    return captured_timeout


Review comment (suggestion):
- Do we have to run the entire agent, or could we just call `GoogleModel.request`?
- Do we have to verify that timeout is passed to the request method, or could we "just" double check that e.g. `provider._client.http_options.timeout` is set? (That specific field probably won't work, but you get the idea). I'd prefer to do less mocking.

Code after:
                ToolReturnPart(tool_name='final_result', content='Final result processed.', tool_call_id='test_id'),
                UserPromptPart(content="What's 2 + 2?"),
            ]
        )
    ]

    _, contents = await m._map_messages(messages, ModelRequestParameters())  # pyright: ignore[reportPrivateUsage]

    assert contents == snapshot(
        [
            {
                'role': 'user',
                'parts': [
                    {
                        'function_response': {
                            'name': 'final_result',
                            'response': {'return_value': 'Final result processed.'},
                            'id': 'test_id',
                        }
                    }
                ],
            },
            {
                'role': 'user',
                'parts': [
                    {
                        'text': "What's 2 + 2?",
                    }
                ],
            },
        ]
    )

    # ToolReturn + Retry + UserPrompts
    messages = [
        ModelRequest(
            parts=[
                ToolReturnPart(tool_name='final_result', content='Final result processed.', tool_call_id='test_id_1'),
                RetryPromptPart(content='Tool error occurred', tool_name='another_tool', tool_call_id='test_id_2'),
                UserPromptPart(content="What's 2 + 2?"),
                UserPromptPart(content="What's 3 + 3?"),
            ]
        )
    ]

    _, contents = await m._map_messages(messages, ModelRequestParameters())  # pyright: ignore[reportPrivateUsage]

    assert contents == snapshot(
        [
            {
                'role': 'user',
Repository: pypa/pip
File: src/pip/_internal/utils/filesystem.py (Python)

Code before:
    return os.path.getsize(path)


def format_file_size(path: str) -> str:
    return format_size(file_size(path))


def directory_size(path: str) -> Union[int, float]:
    size = 0.0
    for root, _dirs, files in os.walk(path):
        for filename in files:
            file_path = os.path.join(root, filename)
            size += file_size(file_path)
    return size


def format_directory_size(path: str) -> str:
    return format_size(directory_size(path))


def list_empty_subdirs(path):
    # type: (str) -> List[str]
    """Returns a list of absolute paths of empty directories beneath path."""
    result = []  # type: List[str]
    for root, dirs, _files in os.walk(path, topdown=False):
        result.extend(os.path.join(root, d) for d in dirs)
    return result


Review comment (suggestion):
We can `yield from ...` here and avoid creating a list.

Code after:
        os.chmod(target_file.name, mode, follow_symlinks=False)


def _subdirs_without_generic(
    path: str, predicate: Callable[[str, list[str]], bool]
) -> Generator[Path]:
    """Yields every subdirectory of +path+ that has no files matching the
    predicate under it."""

    directories = []
    excluded = set()

    for root_str, _, filenames in os.walk(Path(path).resolve()):
        root = Path(root_str)
        if predicate(root_str, filenames):
            # This directory should be excluded, so exclude it and all of its
            # parent directories.
            # The last item in root.parents is ".", so we ignore it.
            #
            # Wrapping this in `list()` is only needed for Python 3.9.
            excluded.update(list(root.parents)[:-1])
            excluded.add(root)
        directories.append(root)

    for d in sorted(directories, reverse=True):
        if d not in excluded:
            yield d


def subdirs_without_files(path: str) -> Generator[Path]:
    """Yields every subdirectory of +path+ that has no files under it."""
    return _subdirs_without_generic(path, lambda root, filenames: len(filenames) > 0)


def subdirs_without_wheels(path: str) -> Generator[Path]:
    """Yields every subdirectory of +path+ that has no .whl files under it."""
    return _subdirs_without_generic(
        path, lambda root, filenames: any(x.endswith(".whl") for x in filenames)
    )
Repository: pypa/pip
File: src/pip/_internal/models/wheel.py (Python)

Code before:

        :raises ValueError: If none of the wheel's file tags match one of
            the supported tags.
        """
        return min(tags.index(tag) for tag in self.file_tags if tag in tags)

    def find_most_preferred_tag(self, tags, tag_to_priority):
        # type: (List[Tag], Dict[Tag, int]) -> int
        """Return the priority of the most preferred tag that one of the wheel's file
        tag combinations acheives in the given list of supported tags using the given
        tag_to_priority mapping, where lower priorities are more-preferred.

        This is used in place of support_index_min in some cases in order to avoid
        an expensive linear scan of a large list of tags.

        :param tags: the PEP 425 tags to check the wheel against.
        :param tag_to_priority: a mapping from tag to priority of that tag, where
            lower is more preferred.

        :raises ValueError: If none of the wheel's file tags match one of
            the supported tags.
        """
        return min(tag_to_priority[tag] for tag in self.file_tags if tag in tag_to_priority)

    def supported(self, tags):
        # type: (Union[List[Tag],Dict[Tag, Any],Set[Tag]]) -> bool
        """Return whether the wheel is compatible with one of the given tags.

        :param tags: the PEP 425 tags to check the wheel against.
        """
        for itag in self.file_tags:
            if itag in tags:
                return True
        return False


Review comment (suggestion):
```suggestion
        # type: (Iterable[Tag]) -> bool
```

I believe the original implementation works as-is.

Code after:

        :raises ValueError: If none of the wheel's file tags match one of
            the supported tags.
        """
        return min(tags.index(tag) for tag in self.file_tags if tag in tags)

    def find_most_preferred_tag(self, tags, tag_to_priority):
        # type: (List[Tag], Dict[Tag, int]) -> int
        """Return the priority of the most preferred tag that one of the wheel's file
        tag combinations acheives in the given list of supported tags using the given
        tag_to_priority mapping, where lower priorities are more-preferred.

        This is used in place of support_index_min in some cases in order to avoid
        an expensive linear scan of a large list of tags.

        :param tags: the PEP 425 tags to check the wheel against.
        :param tag_to_priority: a mapping from tag to priority of that tag, where
            lower is more preferred.

        :raises ValueError: If none of the wheel's file tags match one of
            the supported tags.
        """
        return min(tag_to_priority[tag] for tag in self.file_tags if tag in tag_to_priority)

    def supported(self, tags):
        # type: (Iterable[Tag]) -> bool
        """Return whether the wheel is compatible with one of the given tags.

        :param tags: the PEP 425 tags to check the wheel against.
        """
        return not self.file_tags.isdisjoint(tags)
Repository: pypa/pip
File: src/pip/_internal/utils/packaging.py (Python)

Code before:
    elif dist.has_metadata("PKG-INFO"):
        metadata_name = "PKG-INFO"
        metadata = dist.get_metadata(metadata_name)
    else:
        logger.warning("No metadata found in %s", display_path(dist.location))
        metadata = ""

    if metadata is None:
        raise NoneMetadataError(dist, metadata_name)

    feed_parser = FeedParser()
    # The following line errors out if with a "NoneType" TypeError if
    # passed metadata=None.
    feed_parser.feed(metadata)
    return feed_parser.close()


def get_installer(dist: Distribution) -> str:
    if dist.has_metadata("INSTALLER"):
        for line in dist.get_metadata_lines("INSTALLER"):
            if line.strip():
                return line.strip()
    return ""


@functools.lru_cache(maxsize=None)
def get_requirement(req_string: str) -> Requirement:
    """Construct a packaging.Requirement object with caching"""
    # Parsing requirement strings is expensive, and is also expected to happen
    # with a low diversity of different arguments (at least relative the number
    # constructed). This method adds a cache to requirement object creation to
    # minimize repeated parsing of the same string to construct equivalent
    # Requirement objects.
    return Requirement(req_string)


Review comment (performance):
```suggestion
@functools.lru_cache
```

Let's leave the maxsize as a default, or at least, explicitly set it to some reasonable number. I don't think unbounded memory usage is a behaviour we want the cache to have.

Code after:
    elif dist.has_metadata("PKG-INFO"):
        metadata_name = "PKG-INFO"
        metadata = dist.get_metadata(metadata_name)
    else:
        logger.warning("No metadata found in %s", display_path(dist.location))
        metadata = ""

    if metadata is None:
        raise NoneMetadataError(dist, metadata_name)

    feed_parser = FeedParser()
    # The following line errors out if with a "NoneType" TypeError if
    # passed metadata=None.
    feed_parser.feed(metadata)
    return feed_parser.close()


def get_installer(dist: Distribution) -> str:
    if dist.has_metadata("INSTALLER"):
        for line in dist.get_metadata_lines("INSTALLER"):
            if line.strip():
                return line.strip()
    return ""


@functools.lru_cache(maxsize=512)
def get_requirement(req_string: str) -> Requirement:
    """Construct a packaging.Requirement object with caching"""
    # Parsing requirement strings is expensive, and is also expected to happen
    # with a low diversity of different arguments (at least relative the number
    # constructed). This method adds a cache to requirement object creation to
    # minimize repeated parsing of the same string to construct equivalent
    # Requirement objects.
    return Requirement(req_string)
Repository: pypa/pip
File: tests/unit/test_packaging.py (Python)

Code before:
)
def test_check_requires_python(
    version_info: Tuple[int, int, int], requires_python: Optional[str], expected: bool
) -> None:
    actual = check_requires_python(requires_python, version_info)
    assert actual == expected


def test_check_requires_python__invalid() -> None:
    """
    Test an invalid Requires-Python value.
    """
    with pytest.raises(specifiers.InvalidSpecifier):
        check_requires_python("invalid", (3, 6, 5))


def test_get_or_create_caching() -> None:
    """test caching of get_or_create requirement"""
    teststr = "affinegap==1.10"
    from_helper = get_requirement(teststr)
    freshly_made = Requirement(teststr)
    # Requirement doesn't have an equality operator (yet) so test
    # equality of attribute for list of attributes
    for iattr in ["name", "url", "extras", "specifier", "marker"]:
        assert getattr(from_helper, iattr) == getattr(freshly_made, iattr)
    assert not (get_requirement(teststr) is Requirement(teststr))
    assert get_requirement(teststr) is get_requirement(teststr)


Review comment (suggestion):
```suggestion
    assert get_requirement(teststr) is not Requirement(teststr)
```

Code after:
)
def test_check_requires_python(
    version_info: Tuple[int, int, int], requires_python: Optional[str], expected: bool
) -> None:
    actual = check_requires_python(requires_python, version_info)
    assert actual == expected


def test_check_requires_python__invalid() -> None:
    """
    Test an invalid Requires-Python value.
    """
    with pytest.raises(specifiers.InvalidSpecifier):
        check_requires_python("invalid", (3, 6, 5))


def test_get_or_create_caching() -> None:
    """test caching of get_or_create requirement"""
    teststr = "affinegap==1.10"
    from_helper = get_requirement(teststr)
    freshly_made = Requirement(teststr)

    # Requirement doesn't have an equality operator (yet) so test
    # equality of attribute for list of attributes
    for iattr in ["name", "url", "extras", "specifier", "marker"]:
        assert getattr(from_helper, iattr) == getattr(freshly_made, iattr)
    assert get_requirement(teststr) is not Requirement(teststr)
    assert get_requirement(teststr) is get_requirement(teststr)