requests: Make possible to override headers and allow raw data upload.

Signed-off-by: Mirza Kapetanovic <mirza.kapetanovic@gmail.com>
Mirza Kapetanovic 2024-03-07 16:35:29 +01:00
rodzic ffb07dbce5
commit 80b21be6f4
1 zmienionych plików z 37 dodań i 18 usunięć

Wyświetl plik

@ -38,12 +38,15 @@ def request(
url, url,
data=None, data=None,
json=None, json=None,
headers={}, headers=None,
stream=None, stream=None,
auth=None, auth=None,
timeout=None, timeout=None,
parse_headers=True, parse_headers=True,
): ):
if headers is None:
headers = {}
redirect = None # redirection url, None means no redirection redirect = None # redirection url, None means no redirection
chunked_data = data and getattr(data, "__next__", None) and not getattr(data, "__len__", None) chunked_data = data and getattr(data, "__next__", None) and not getattr(data, "__len__", None)
@ -94,33 +97,49 @@ def request(
context.verify_mode = tls.CERT_NONE context.verify_mode = tls.CERT_NONE
s = context.wrap_socket(s, server_hostname=host) s = context.wrap_socket(s, server_hostname=host)
s.write(b"%s /%s HTTP/1.0\r\n" % (method, path)) s.write(b"%s /%s HTTP/1.0\r\n" % (method, path))
if "Host" not in headers: if "Host" not in headers:
s.write(b"Host: %s\r\n" % host) headers["Host"] = host
if json is not None:
assert data is None
import ujson
data = ujson.dumps(json)
if "Content-Type" not in headers:
headers["Content-Type"] = "application/json"
if data:
if chunked_data:
if "Transfer-Encoding" not in headers and "Content-Length" not in headers:
headers["Transfer-Encoding"] = "chunked"
elif "Content-Length" not in headers:
headers["Content-Length"] = str(len(data))
if "Connection" not in headers:
headers["Connection"] = "close"
# Iterate over keys to avoid tuple alloc # Iterate over keys to avoid tuple alloc
for k in headers: for k in headers:
s.write(k) s.write(k)
s.write(b": ") s.write(b": ")
s.write(headers[k]) s.write(headers[k])
s.write(b"\r\n") s.write(b"\r\n")
if json is not None:
assert data is None
import ujson
data = ujson.dumps(json) s.write(b"\r\n")
s.write(b"Content-Type: application/json\r\n")
if data: if data:
if chunked_data: if chunked_data:
s.write(b"Transfer-Encoding: chunked\r\n") if headers.get("Transfer-Encoding", None) == "chunked":
else: for chunk in data:
s.write(b"Content-Length: %d\r\n" % len(data)) s.write(b"%x\r\n" % len(chunk))
s.write(b"Connection: close\r\n\r\n") s.write(chunk)
if data: s.write(b"\r\n")
if chunked_data: s.write("0\r\n\r\n")
for chunk in data: else:
s.write(b"%x\r\n" % len(chunk)) for chunk in data:
s.write(chunk) s.write(chunk)
s.write(b"\r\n")
s.write("0\r\n\r\n")
else: else:
s.write(data) s.write(data)