Add sentinel type for _FAILEDTELL
MODIFY
src/urllib3/connectionpool.py
src/urllib3/connectionpool.py
@@ -41,7 +41,7 @@ from .exceptions import (
from .response import BaseHTTPResponse, HTTPResponse
from .util.connection import is_connection_dropped
from .util.proxy import connection_requires_http_tunnel
-from .util.request import set_file_position
+from .util.request import _TYPE_BODY_POSITION, set_file_position
from .util.response import assert_header_parsing
from .util.retry import Retry
from .util.ssl_match_hostname import CertificateError
@@ -547,7 +547,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
pool_timeout: Optional[int] = None,
release_conn: Optional[bool] = None,
chunked: bool = False,
- body_pos: Optional[Union[int, object]] = None,
+ body_pos: Optional[_TYPE_BODY_POSITION] = None,
**response_kw: Any,
) -> BaseHTTPResponse:
"""

MODIFY
src/urllib3/util/request.py
src/urllib3/util/request.py
from base64 import b64encode
-from typing import IO, Any, AnyStr, Dict, List, Optional, Union
+from enum import Enum
+from typing import IO, TYPE_CHECKING, Any, AnyStr, Dict, List, Optional, Union
from ..exceptions import UnrewindableBodyError
+if TYPE_CHECKING:
+ from typing_extensions import Final
+
# Pass as a value within ``headers`` to skip
# emitting some HTTP headers that are added automatically.
# The only headers that are supported are ``Accept-Encoding``,
@@ -21,7 +25,14 @@ except ImportError:
else:
ACCEPT_ENCODING += ",br"
-_FAILEDTELL = object()
+
+class _TYPE_FAILEDTELL(Enum):
+ token = 0
+
+
+_FAILEDTELL: "Final[_TYPE_FAILEDTELL]" = _TYPE_FAILEDTELL.token
+
+_TYPE_BODY_POSITION = Union[int, _TYPE_FAILEDTELL]
def make_headers(
@@ -104,8 +115,8 @@ def make_headers(
def set_file_position(
- body: Any, pos: Optional[Union[int, object]]
-) -> Optional[Union[int, object]]:
+ body: Any, pos: Optional[_TYPE_BODY_POSITION]
+) -> Optional[_TYPE_BODY_POSITION]:
"""
If a position is provided, move file to that point.
Otherwise, we'll attempt to record a position for future use.
@@ -123,7 +134,7 @@ def set_file_position(
return pos
-def rewind_body(body: IO[AnyStr], body_pos: Optional[Union[int, object]]) -> None:
+def rewind_body(body: IO[AnyStr], body_pos: _TYPE_BODY_POSITION) -> None:
"""
Attempt to rewind body to a certain position.
Primarily used for request redirects and retries.

MODIFY
test/test_util.py
test/test_util.py
@@ -530,9 +530,9 @@ class TestUtil:
# Pass non-integer position
with pytest.raises(ValueError):
- rewind_body(body, body_pos=None)
+ rewind_body(body, body_pos=None) # type: ignore[arg-type]
with pytest.raises(ValueError):
- rewind_body(body, body_pos=object())
+ rewind_body(body, body_pos=object()) # type: ignore[arg-type]
def test_rewind_body_failed_seek(self) -> None:
class BadSeek(io.StringIO):

Add sentinel type for _FAILEDTELL (#2519)