typing.List[typing.Tuple[typing.AnyStr, typing.AnyStr]],
]
-RequestData = typing.Union[bytes, typing.AsyncIterator[bytes]]
+RequestData = typing.Union[dict, bytes, typing.AsyncIterator[bytes]]
ResponseContent = typing.Union[bytes, typing.AsyncIterator[bytes]]
return hash((self.is_ssl, self.host, self.port))
-class QueryParams(typing.Mapping):
+class QueryParams(typing.Mapping[str, str]):
"""
URL query parameters, as a multi-dict.
"""
):
self.method = method.upper()
self.url = URL(url, query_params=query_params)
+ self.headers = Headers(headers)
+
if isinstance(data, bytes):
self.is_streaming = False
self.content = data
+ elif isinstance(data, dict):
+ self.is_streaming = False
+ self.content = urlencode(data, doseq=True).encode("utf-8")
+ self.headers["Content-Type"] = "application/x-www-form-urlencoded"
else:
self.is_streaming = True
self.content_aiter = data
- self.headers = Headers(headers)
async def read(self) -> bytes:
"""
)
+def test_url_encoded_data():
+ request = httpcore.Request("POST", "http://example.org", data={"test": "123"})
+ request.prepare()
+ assert request.headers == httpcore.Headers(
+ [
+ (b"host", b"example.org"),
+ (b"content-length", b"8"),
+ (b"accept-encoding", b"deflate, gzip, br"),
+ (b"content-type", b"application/x-www-form-urlencoded"),
+ ]
+ )
+ assert request.content == b"test=123"
+
+
def test_transfer_encoding_header():
async def streaming_body(data):
yield data # pragma: nocover
data = streaming_body(b"test 123")
headers = [(b"content-length", b"8")]
- request = httpcore.Request(
- "POST", "http://example.org", data=data, headers=headers
- )
+ request = httpcore.Request("POST", "http://example.org", data=data, headers=headers)
request.prepare()
assert request.headers == httpcore.Headers(
[