Added `preload_content`, `retries`, `redirect` and `timeout` arguments to urllib3.request()
MODIFY
CONTRIBUTORS.txt
CONTRIBUTORS.txt
@@ -321,5 +321,8 @@ In chronological order:
* Bastian Venthur <https://venthur.de>
* fixed all remaining coverage checks for raise statements
+* Ubdussamad <ubdussamad@gmail.com>
+ * Added multiple parameters to the top-level request method.
+
* [Your name or handle] <[email or website]>
* [Brief summary of your changes]

MODIFY
src/urllib3/__init__.py
src/urllib3/__init__.py
@@ -6,7 +6,7 @@ Python HTTP library with thread-safe connection pooling, file post support, user
import logging
import warnings
from logging import NullHandler
-from typing import Mapping, Optional, Type
+from typing import Mapping, Optional, Type, Union
from . import exceptions
from ._collections import HTTPHeaderDict
@@ -94,6 +94,10 @@ def request(
body: Optional[HTTPBody] = None,
fields: Optional[_TYPE_FIELDS] = None,
headers: Optional[Mapping[str, str]] = None,
+ preload_content: Optional[bool] = True,
+ redirect: Optional[bool] = True,
+ retries: Optional[Union[Retry, bool, int]] = None,
+ timeout: Optional[Union[Timeout, float, int]] = 3,
) -> BaseHTTPResponse:
"""
A convenience, top-level request method. It uses a module-global ``PoolManager`` instance.
@@ -102,4 +106,14 @@ def request(
The method does not accept low-level ``**urlopen_kw`` keyword arguments.
"""
- return _DEFAULT_POOL.request(method, url, body=body, fields=fields, headers=headers)
+ return _DEFAULT_POOL.request(
+ method,
+ url,
+ body=body,
+ fields=fields,
+ headers=headers,
+ preload_content=preload_content,
+ redirect=redirect,
+ retries=retries,
+ timeout=timeout,
+ )

MODIFY
test/with_dummyserver/test_poolmanager.py
test/with_dummyserver/test_poolmanager.py
import json
from test import LONG_TIMEOUT
+from unittest import mock
import pytest
from dummyserver.server import HAS_IPV6
from dummyserver.testcase import HTTPDummyServerTestCase, IPv6HTTPDummyServerTestCase
-from urllib3 import request
+from urllib3 import HTTPResponse, request
from urllib3.connectionpool import port_by_scheme
from urllib3.exceptions import MaxRetryError, URLSchemeUnknown
from urllib3.poolmanager import PoolManager
@@ -378,6 +379,60 @@ class TestPoolManager(HTTPDummyServerTestCase):
assert r.status == 200
assert r.data == b"test"
+ def test_top_level_request_with_preload_content(self):
+ r = request("GET", f"{self.base_url}/echo", preload_content=False)
+ assert r.status == 200
+ assert r.connection is not None
+ r.data
+ assert r.connection is None
+
+ def test_top_level_request_with_redirect(self):
+ r = request(
+ "GET",
+ f"{self.base_url}/redirect",
+ fields={"target": f"{self.base_url}/"},
+ redirect=False,
+ )
+
+ assert r.status == 303
+
+ r = request(
+ "GET",
+ f"{self.base_url}/redirect",
+ fields={"target": f"{self.base_url}/"},
+ redirect=True,
+ )
+
+ assert r.status == 200
+ assert r.data == b"Dummy server!"
+
+ def test_top_level_request_with_retries(self):
+ r = request("GET", f"{self.base_url}/redirect", retries=False)
+ assert r.status == 303
+
+ r = request("GET", f"{self.base_url}/redirect", retries=3)
+ assert r.status == 200
+
+ def test_top_level_request_with_timeout(self):
+ with mock.patch("urllib3.poolmanager.RequestMethods.request") as mockRequest:
+ mockRequest.return_value = HTTPResponse(status=200)
+
+ r = request("GET", f"{self.base_url}/redirect", timeout=2.5)
+
+ assert r.status == 200
+
+ mockRequest.assert_called_with(
+ "GET",
+ f"{self.base_url}/redirect",
+ body=None,
+ fields=None,
+ headers=None,
+ preload_content=True,
+ redirect=True,
+ retries=None,
+ timeout=2.5,
+ )
+
@pytest.mark.skipif(not HAS_IPV6, reason="IPv6 is not supported on this system")
class TestIPv6PoolManager(IPv6HTTPDummyServerTestCase):

Added `preload_content`, `retries`, `redirect` and `timeout` arguments to urllib3.request()