ronantakizawa/codereview-bench huggingface dataset:
Repository: encode/httpx
File: httpx/_utils.py (Python)

Code before:
) -> bytes:
    """
    Coerce str/bytes into a strictly byte-wise HTTP header key.
    """
    if isinstance(value, bytes):
        bytes_value = value
    else:
        bytes_value = value.encode(encoding or "ascii")

    return bytes_value.lower() if lower else bytes_value


def normalize_header_value(
    value: typing.Union[str, bytes], encoding: str = None
) -> bytes:
    """
    Coerce str/bytes into a strictly byte-wise HTTP header value.
    """
    if isinstance(value, bytes):
        return value
    return value.encode(encoding or "ascii")


def primitive_value_to_str(value: "PrimitiveData") -> str:
    """
    Coerce a primitive data type into a string value for query params.

    Note that we prefer JSON-style 'true'/'false' for boolean values here.
    """
    if value is True:
        return "true"
    elif value is False:
        return "false"
    elif value is None:
        return ""
    return str(value)


def is_known_encoding(encoding: str) -> bool:
    """
    Return `True` if `encoding` is a known codec.
    """
    try:
        codecs.lookup(encoding)
    except LookupError:
        return False
    return True


def format_form_param(name: str, value: typing.Union[str, bytes]) -> bytes:
    """

Review comment (suggestion):
(Nit)

```suggestion
    Coerce a primitive data type into a string value.
```

Code after:
) -> bytes:
    """
    Coerce str/bytes into a strictly byte-wise HTTP header key.
    """
    if isinstance(value, bytes):
        bytes_value = value
    else:
        bytes_value = value.encode(encoding or "ascii")

    return bytes_value.lower() if lower else bytes_value


def normalize_header_value(
    value: typing.Union[str, bytes], encoding: str = None
) -> bytes:
    """
    Coerce str/bytes into a strictly byte-wise HTTP header value.
    """
    if isinstance(value, bytes):
        return value
    return value.encode(encoding or "ascii")


def primitive_value_to_str(value: "PrimitiveData") -> str:
    """
    Coerce a primitive data type into a string value.

    Note that we prefer JSON-style 'true'/'false' for boolean values here.
    """
    if value is True:
        return "true"
    elif value is False:
        return "false"
    elif value is None:
        return ""
    return str(value)


def is_known_encoding(encoding: str) -> bool:
    """
    Return `True` if `encoding` is a known codec.
    """
    try:
        codecs.lookup(encoding)
    except LookupError:
        return False
    return True


def format_form_param(name: str, value: typing.Union[str, bytes]) -> bytes:
    """
Repository: encode/httpx
File: httpx/_transports/base.py (Python)

Code before:
import typing
from types import TracebackType

T = typing.TypeVar("T", bound="BaseTransport")
A = typing.TypeVar("A", bound="AsyncBaseTransport")


class SyncByteStream:
    def __iter__(self) -> typing.Iterator[bytes]:
        raise NotImplementedError(
            "The '__iter__' method must be implemented."
        )  # pragma: nocover
        yield b""  # pragma: nocover

    def close(self) -> None:
        """
        Subclasses can override this method to release any network resources
        after a request/response cycle is complete.

        Streaming cases should call use a `try...finally` block to ensure that
        the stream `close()` method is always called.

        Example:

            status_code, headers, stream, extensions = transport.handle_request(...)
            try:
                ...
            finally:
                stream.close()
        """

    def read(self) -> bytes:
        """
        Simple cases can use `.read()` as a convience method for consuming
        the entire stream and then closing it.

        Example:

            status_code, headers, stream, extensions = transport.handle_request(...)
            body = stream.read()
        """
        try:
            return b"".join([part for part in self])
        finally:
            self.close()

Review comment (suggestion):
```suggestion
        Streaming cases should use a `try...finally` block to ensure that
```

Code after:
import typing
from types import TracebackType

T = typing.TypeVar("T", bound="BaseTransport")
A = typing.TypeVar("A", bound="AsyncBaseTransport")


class SyncByteStream:
    def __iter__(self) -> typing.Iterator[bytes]:
        raise NotImplementedError(
            "The '__iter__' method must be implemented."
        )  # pragma: nocover
        yield b""  # pragma: nocover

    def close(self) -> None:
        """
        Subclasses can override this method to release any network resources
        after a request/response cycle is complete.

        Streaming cases should use a `try...finally` block to ensure that
        the stream `close()` method is always called.

        Example:

            status_code, headers, stream, extensions = transport.handle_request(...)
            try:
                ...
            finally:
                stream.close()
        """

    def read(self) -> bytes:
        """
        Simple cases can use `.read()` as a convience method for consuming
        the entire stream and then closing it.

        Example:

            status_code, headers, stream, extensions = transport.handle_request(...)
            body = stream.read()
        """
        try:
            return b"".join([part for part in self])
        finally:
            self.close()
Repository: encode/httpx
File: httpx/_transports/default.py (Python)

Code before:
        stream: AsyncByteStream,
        extensions: dict,
    ) -> typing.Tuple[
        int, typing.List[typing.Tuple[bytes, bytes]], AsyncByteStream, dict
    ]:
        with map_httpcore_exceptions():
            status_code, headers, byte_stream, extensions = await self._pool.arequest(
                method=method,
                url=url,
                headers=headers,
                stream=httpcore.AsyncIteratorByteStream(stream.__aiter__()),
                ext=extensions,
            )

        class ResponseStream(AsyncByteStream):
            def __init__(self, httpcore_stream: httpcore.AsyncByteStream):
                self._httpcore_stream = httpcore_stream

            async def __aiter__(self) -> typing.AsyncIterator[bytes]:
                with map_httpcore_exceptions():
                    async for part in self._httpcore_stream:
                        yield part

            async def aclose(self) -> None:
                with map_httpcore_exceptions():
                    await byte_stream.aclose()

        ensure_http_version_reason_phrase_as_bytes(extensions)
        stream = ResponseStream(byte_stream)

        return status_code, headers, stream, extensions

    async def aclose(self) -> None:
        await self._pool.aclose()


Review comment (suggestion):
```suggestion
                    await self._httpcore_stream.aclose()
```

Code after:
        return self

    async def __aexit__(
        self,
        exc_type: typing.Type[BaseException] = None,
        exc_value: BaseException = None,
        traceback: TracebackType = None,
    ) -> None:
        await self._pool.__aexit__(exc_type, exc_value, traceback)

    async def handle_async_request(
        self,
        method: bytes,
        url: typing.Tuple[bytes, bytes, typing.Optional[int], bytes],
        headers: typing.List[typing.Tuple[bytes, bytes]],
        stream: AsyncByteStream,
        extensions: dict,
    ) -> typing.Tuple[
        int, typing.List[typing.Tuple[bytes, bytes]], AsyncByteStream, dict
    ]:
        with map_httpcore_exceptions():
            status_code, headers, byte_stream, extensions = await self._pool.arequest(
                method=method,
                url=url,
                headers=headers,
                stream=httpcore.AsyncIteratorByteStream(stream.__aiter__()),
                ext=extensions,
            )

        ensure_http_version_reason_phrase_as_bytes(extensions)
        stream = AsyncResponseStream(byte_stream)

        return status_code, headers, stream, extensions

    async def aclose(self) -> None:
        await self._pool.aclose()
Repository: encode/httpx
File: httpx/_content.py (Python)

Code before:
import inspect
import warnings
from json import dumps as json_dumps
from typing import (
    Any,
    AsyncIterable,
    AsyncIterator,
    Dict,
    Iterable,
    Iterator,
    Tuple,
    Union,
)
from urllib.parse import urlencode

from ._exceptions import ResponseClosed, StreamConsumed
from ._multipart import MultipartStream
from ._transports.base import AsyncByteStream, SyncByteStream
from ._types import RequestContent, RequestData, RequestFiles, ResponseContent
from ._utils import primitive_value_to_str


class ByteStream(AsyncByteStream, SyncByteStream):
    def __init__(self, stream: bytes) -> None:
        self._stream = stream

    def __iter__(self) -> Iterator[bytes]:
        yield self._stream

    async def __aiter__(self) -> AsyncIterator[bytes]:
        yield self._stream


class IteratorByteStream(SyncByteStream):
    def __init__(self, stream: Iterable[bytes]):
        self._stream = stream
        self._is_stream_consumed = False
        self._is_generator = inspect.isgenerator(stream)

    def __iter__(self) -> Iterator[bytes]:
        if self._is_stream_consumed and self._is_generator:

Review comment (suggestion):
```suggestion
from ._exceptions import StreamClosed, StreamConsumed
```

Code after:
import inspect
import warnings
from json import dumps as json_dumps
from typing import (
    Any,
    AsyncIterable,
    AsyncIterator,
    Dict,
    Iterable,
    Iterator,
    Tuple,
    Union,
)
from urllib.parse import urlencode

from ._exceptions import StreamClosed, StreamConsumed
from ._multipart import MultipartStream
from ._transports.base import AsyncByteStream, SyncByteStream
from ._types import RequestContent, RequestData, RequestFiles, ResponseContent
from ._utils import primitive_value_to_str


class ByteStream(AsyncByteStream, SyncByteStream):
    def __init__(self, stream: bytes) -> None:
        self._stream = stream

    def __iter__(self) -> Iterator[bytes]:
        yield self._stream

    async def __aiter__(self) -> AsyncIterator[bytes]:
        yield self._stream


class IteratorByteStream(SyncByteStream):
    def __init__(self, stream: Iterable[bytes]):
        self._stream = stream
        self._is_stream_consumed = False
        self._is_generator = inspect.isgenerator(stream)

    def __iter__(self) -> Iterator[bytes]:
        if self._is_stream_consumed and self._is_generator:
Repository: encode/httpx
File: tests/models/test_requests.py (Python)

Code before:
    async def streaming_body(data):
        yield data

    data = streaming_body(b"test 123")
    request = httpx.Request("POST", "http://example.org", content=data)
    pickle_request = pickle.loads(pickle.dumps(request))
    with pytest.raises(httpx.RequestNotRead):
        pickle_request.content
    with pytest.raises(httpx.ResponseClosed):  # TODO: StreamClosed
        await pickle_request.aread()

    request = httpx.Request("POST", "http://example.org", content=data)
    await request.aread()
    pickle_request = pickle.loads(pickle.dumps(request))
    assert pickle_request.content == b"test 123"


def test_request_generator_content_picklable():
    def content():
        yield b"test 123"  # pragma: nocover

    request = httpx.Request("POST", "http://example.org", content=content())
    pickle_request = pickle.loads(pickle.dumps(request))
    with pytest.raises(httpx.RequestNotRead):
        pickle_request.content
    with pytest.raises(httpx.ResponseClosed):  # TODO: StreamClosed
        pickle_request.read()

    request = httpx.Request("POST", "http://example.org", content=content())
    request.read()
    pickle_request = pickle.loads(pickle.dumps(request))
    assert pickle_request.content == b"test 123"


Review comment (suggestion):
```suggestion
    with pytest.raises(https.StreamClosed):
```

Code after:
    async def streaming_body(data):
        yield data

    data = streaming_body(b"test 123")
    request = httpx.Request("POST", "http://example.org", content=data)
    pickle_request = pickle.loads(pickle.dumps(request))
    with pytest.raises(httpx.RequestNotRead):
        pickle_request.content
    with pytest.raises(httpx.StreamClosed):
        await pickle_request.aread()

    request = httpx.Request("POST", "http://example.org", content=data)
    await request.aread()
    pickle_request = pickle.loads(pickle.dumps(request))
    assert pickle_request.content == b"test 123"


def test_request_generator_content_picklable():
    def content():
        yield b"test 123"  # pragma: nocover

    request = httpx.Request("POST", "http://example.org", content=content())
    pickle_request = pickle.loads(pickle.dumps(request))
    with pytest.raises(httpx.RequestNotRead):
        pickle_request.content
    with pytest.raises(httpx.StreamClosed):
        pickle_request.read()

    request = httpx.Request("POST", "http://example.org", content=content())
    request.read()
    pickle_request = pickle.loads(pickle.dumps(request))
    assert pickle_request.content == b"test 123"
Repository: encode/httpx
File: tests/models/test_requests.py (Python)

Code before:
    async def streaming_body(data):
        yield data

    data = streaming_body(b"test 123")
    request = httpx.Request("POST", "http://example.org", content=data)
    pickle_request = pickle.loads(pickle.dumps(request))
    with pytest.raises(httpx.RequestNotRead):
        pickle_request.content
    with pytest.raises(httpx.StreamClosed):
        await pickle_request.aread()

    request = httpx.Request("POST", "http://example.org", content=data)
    await request.aread()
    pickle_request = pickle.loads(pickle.dumps(request))
    assert pickle_request.content == b"test 123"


def test_request_generator_content_picklable():
    def content():
        yield b"test 123"  # pragma: nocover

    request = httpx.Request("POST", "http://example.org", content=content())
    pickle_request = pickle.loads(pickle.dumps(request))
    with pytest.raises(httpx.RequestNotRead):
        pickle_request.content
    with pytest.raises(https.StreamClosed):
        pickle_request.read()

    request = httpx.Request("POST", "http://example.org", content=content())
    request.read()
    pickle_request = pickle.loads(pickle.dumps(request))
    assert pickle_request.content == b"test 123"


Review comment (suggestion):
```suggestion
    with pytest.raises(httpx.StreamClosed):
```

Code after:
    async def streaming_body(data):
        yield data

    data = streaming_body(b"test 123")
    request = httpx.Request("POST", "http://example.org", content=data)
    pickle_request = pickle.loads(pickle.dumps(request))
    with pytest.raises(httpx.RequestNotRead):
        pickle_request.content
    with pytest.raises(httpx.StreamClosed):
        await pickle_request.aread()

    request = httpx.Request("POST", "http://example.org", content=data)
    await request.aread()
    pickle_request = pickle.loads(pickle.dumps(request))
    assert pickle_request.content == b"test 123"


def test_request_generator_content_picklable():
    def content():
        yield b"test 123"  # pragma: nocover

    request = httpx.Request("POST", "http://example.org", content=content())
    pickle_request = pickle.loads(pickle.dumps(request))
    with pytest.raises(httpx.RequestNotRead):
        pickle_request.content
    with pytest.raises(httpx.StreamClosed):
        pickle_request.read()

    request = httpx.Request("POST", "http://example.org", content=content())
    request.read()
    pickle_request = pickle.loads(pickle.dumps(request))
    assert pickle_request.content == b"test 123"
Repository: encode/httpx
File: httpx/_models.py (Python)

Code before:
        assert url.raw_host == b"xn--fiqs8s.icom.museum"

        url = httpx.URL("https://[::ffff:192.168.0.1]")
        assert url.raw_host == b"::ffff:192.168.0.1"
        """
        host: str = self._uri_reference.host or ""

        if host and ":" in host and host[0] == "[":
            # it's an IPv6 address
            host = host.lstrip("[").rstrip("]")

        return host.encode("ascii")

    @property
    def port(self) -> typing.Optional[int]:
        """
        The URL port as an integer.
        """
        port = self._uri_reference.port
        return int(port) if port else None

    @property
    def netloc(self) -> bytes:
        """
        Either `<host>` or `<host>:<port>` as bytes.
        Always normlized to lowercase, and IDNA encoded.
        """
        host = self._uri_reference.host or ""
        port = self._uri_reference.port
        netloc = host.encode("ascii")
        if port:
            netloc = netloc + b":" + str(port).encode("ascii")
        return netloc

    @property
    def path(self) -> str:
        """
        The URL path as a string. Excluding the query string, and URL decoded.

        For example:

        url = httpx.URL("https://example.com/pa%20th")
        assert url.path == "/pa th"
        """
        path = self._uri_reference.path or "/"
        return unquote(path)

    @property
    def query(self) -> bytes:
        """
        The URL query string, as raw bytes, excluding the leading b"?".

Review comment (suggestion):
```suggestion
        Always normalized to lowercase, and IDNA encoded.
```

Code after:
        assert url.raw_host == b"xn--fiqs8s.icom.museum"

        url = httpx.URL("https://[::ffff:192.168.0.1]")
        assert url.raw_host == b"::ffff:192.168.0.1"
        """
        host: str = self._uri_reference.host or ""

        if host and ":" in host and host[0] == "[":
            # it's an IPv6 address
            host = host.lstrip("[").rstrip("]")

        return host.encode("ascii")

    @property
    def port(self) -> typing.Optional[int]:
        """
        The URL port as an integer.
        """
        port = self._uri_reference.port
        return int(port) if port else None

    @property
    def netloc(self) -> bytes:
        """
        Either `<host>` or `<host>:<port>` as bytes.
        Always normalized to lowercase, and IDNA encoded.
        """
        host = self._uri_reference.host or ""
        port = self._uri_reference.port
        netloc = host.encode("ascii")
        if port:
            netloc = netloc + b":" + str(port).encode("ascii")
        return netloc

    @property
    def path(self) -> str:
        """
        The URL path as a string. Excluding the query string, and URL decoded.

        For example:

        url = httpx.URL("https://example.com/pa%20th")
        assert url.path == "/pa th"
        """
        path = self._uri_reference.path or "/"
        return unquote(path)

    @property
    def query(self) -> bytes:
        """
        The URL query string, as raw bytes, excluding the leading b"?".
Repository: encode/httpx
File: httpx/_compat.py (Python)

Code before:
"""
The _compat module is used for code which requires branching between different
Python environments. It is excluded from the code coverage checks.
"""
import ssl
import sys

# `contextlib.asynccontextmanager` exists from Python 3.7 onwards.
# For 3.6 we require the `async_generator` package for a backported version.
try:
    from contextlib import asynccontextmanager  # type: ignore
except ImportError:
    from async_generator import asynccontextmanager  # type: ignore # noqa


if sys.version_info >= (3, 10) or (
    sys.version_info >= (3, 7) and ssl.OPENSSL_VERSION_INFO >= (1, 1, 0, 7)
):
    # The OP_NO_SSL* and OP_NO_TLS* become deprecated in favor of
    # 'SSLContext.minimum_version' from Python 3.7 onwards, however
    # this attribute is not available unless the ssl module is compiled
    # with OpenSSL 1.1.0g or newer.
    def set_minimum_tls_version_1_2(context: ssl.SSLContext) -> None:
        context.minimum_version = ssl.TLSVersion.TLSv1_2


else:

    def set_minimum_tls_version_1_2(context: ssl.SSLContext) -> None:
        context.options |= ssl.OP_NO_SSLv2
        context.options |= ssl.OP_NO_SSLv3
        context.options |= ssl.OP_NO_TLSv1
        context.options |= ssl.OP_NO_TLSv1_1


Review comment (suggestion):
```suggestion
    # with OpenSSL 1.1.0g or newer.
    # https://docs.python.org/3.10/library/ssl.html#ssl.SSLContext.minimum_version
    # https://docs.python.org/3.7/library/ssl.html#ssl.SSLContext.minimum_version
```

Code after:
"""
The _compat module is used for code which requires branching between different
Python environments. It is excluded from the code coverage checks.
"""
import ssl
import sys

# `contextlib.asynccontextmanager` exists from Python 3.7 onwards.
# For 3.6 we require the `async_generator` package for a backported version.
try:
    from contextlib import asynccontextmanager  # type: ignore
except ImportError:
    from async_generator import asynccontextmanager  # type: ignore # noqa


def set_minimum_tls_version_1_2(context: ssl.SSLContext) -> None:
    if sys.version_info >= (3, 10) or (
        sys.version_info >= (3, 7) and ssl.OPENSSL_VERSION_INFO >= (1, 1, 0, 7)
    ):
        # The OP_NO_SSL* and OP_NO_TLS* become deprecated in favor of
        # 'SSLContext.minimum_version' from Python 3.7 onwards, however
        # this attribute is not available unless the ssl module is compiled
        # with OpenSSL 1.1.0g or newer.
        # https://docs.python.org/3.10/library/ssl.html#ssl.SSLContext.minimum_version
        # https://docs.python.org/3.7/library/ssl.html#ssl.SSLContext.minimum_version
        context.minimum_version = ssl.TLSVersion.TLSv1_2
    else:
        # If 'minimum_version' isn't available, we configure these options with
        # the older deprecated variants.
        context.options |= ssl.OP_NO_SSLv2
        context.options |= ssl.OP_NO_SSLv3
        context.options |= ssl.OP_NO_TLSv1
        context.options |= ssl.OP_NO_TLSv1_1