Added body parameter to top-level request() and RequestMethods.request()
MODIFY
CONTRIBUTORS.txt
CONTRIBUTORS.txt
@@ -315,5 +315,8 @@ In chronological order:
* Ubdussamad <ubdussamad@gmail.com>
* Added HTTPHeaderDict to top level objects and added relevant usage documentation for the same.
+* Bastian Venthur <https://venthur.de>
+ * Added body-parameter to top-level request() and RequestMethods.request
+
* [Your name or handle] <[email or website]>
* [Brief summary of your changes]

MODIFY
src/urllib3/__init__.py
src/urllib3/__init__.py
@@ -86,7 +86,7 @@ def disable_warnings(category=exceptions.HTTPWarning):
_DEFAULT_POOL = PoolManager()
-def request(method, url, fields=None, headers=None):
+def request(method, url, body=None, fields=None, headers=None):
"""
A convenience, top-level request method. It uses a module-global ``PoolManager`` instance.
Therefore, its side effects could be shared across dependencies relying on it.
@@ -94,4 +94,4 @@ def request(method, url, fields=None, headers=None):
The method does not accept low-level ``**urlopen_kw`` keyword arguments.
"""
- return _DEFAULT_POOL.request(method, url, fields=fields, headers=headers)
+ return _DEFAULT_POOL.request(method, url, body=body, fields=fields, headers=headers)

MODIFY
src/urllib3/request.py
src/urllib3/request.py
@@ -65,6 +65,7 @@ class RequestMethods:
self,
method: str,
url: str,
+ body: Optional[HTTPBody] = None,
fields: Optional[_TYPE_FIELDS] = None,
headers: Optional[Mapping[str, str]] = None,
**urlopen_kw: Any,
@@ -83,6 +84,9 @@ class RequestMethods:
urlopen_kw["request_url"] = url
+ if body is not None:
+ urlopen_kw["body"] = body
+
if method in self._encode_url_methods:
return self.request_encode_url(
method,

MODIFY
test/with_dummyserver/test_poolmanager.py
test/with_dummyserver/test_poolmanager.py
@@ -334,6 +334,11 @@ class TestPoolManager(HTTPDummyServerTestCase):
assert returned_headers.get("Foo") is None
assert returned_headers.get("Baz") == "quux"
+ def test_body(self):
+ with PoolManager() as http:
+ r = http.request("POST", f"{self.base_url}/echo", body=b"test")
+ assert r.data == b"test"
+
def test_http_with_ssl_keywords(self):
with PoolManager(ca_certs="REQUIRED") as http:
r = http.request("GET", f"http://{self.host}:{self.port}/")
@@ -368,6 +373,11 @@ class TestPoolManager(HTTPDummyServerTestCase):
assert r.status == 200
assert r.data == b"Dummy server!"
+ def test_top_level_request_with_body(self):
+ r = request("POST", f"{self.base_url}/echo", body=b"test")
+ assert r.status == 200
+ assert r.data == b"test"
+
@pytest.mark.skipif(not HAS_IPV6, reason="IPv6 is not supported on this system")
class TestIPv6PoolManager(IPv6HTTPDummyServerTestCase):

Added body parameter to top-level request() and RequestMethods.request()