assert timeout is None or isinstance(timeout, TimeoutConfig)
# Start sending the request.
- method = request.method.encode('ascii')
- target = request.url.full_path.encode('ascii')
+ method = request.method.encode("ascii")
+ target = request.url.full_path.encode("ascii")
headers = request.headers.raw
event = h11.Request(method=method, target=target, headers=headers)
await self._send_event(event, timeout)
headers = []
for k, v in event.headers:
if k == b":status":
- status_code = int(v.decode('ascii', errors='ignore'))
+ status_code = int(v.decode("ascii", errors="ignore"))
elif not k.startswith(b":"):
headers.append((k, v))
async def send_headers(self, request: Request, timeout: OptionalTimeout) -> int:
stream_id = self.h2_state.get_next_available_stream_id()
headers = [
- (b":method", request.method.encode('ascii')),
- (b":authority", request.url.authority.encode('ascii')),
- (b":scheme", request.url.scheme.encode('ascii')),
- (b":path", request.url.full_path.encode('ascii')),
+ (b":method", request.method.encode("ascii")),
+ (b":authority", request.url.authority.encode("ascii")),
+ (b":scheme", request.url.scheme.encode("ascii")),
+ (b":path", request.url.full_path.encode("ascii")),
] + request.headers.raw
self.h2_state.send_headers(stream_id, headers)
data_to_send = self.h2_state.data_to_send()
if content_type is None:
return None
+ parsed = cgi.parse_header(content_type)
+ media_type, params = parsed[0], parsed[-1]
+ if "charset" in params:
+ return params["charset"].strip("'\"")
+
# RFC 2616 specifies that 'iso-8859-1' should be used as the default
# for 'text/*' media types, if no charset is provided.
# See: https://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html#sec3.7.1
- parsed = cgi.parse_header(content_type)
- media_type, info = parsed[0], parsed[-1]
- default = "iso-8859-1" if media_type.startswith("text/") else None
- return info.get("charset", default)
+ if media_type.startswith("text/"):
+ return "iso-8859-1"
+
+ return None
@property
def apparent_encoding(self) -> typing.Optional[str]:
assert url.path == "/path/to/somewhere"
assert url.query == "abc=123"
assert url.fragment == "anchor"
- assert repr(url) == "URL('https://example.org:123/path/to/somewhere?abc=123#anchor')"
+ assert (
+ repr(url) == "URL('https://example.org:123/path/to/somewhere?abc=123#anchor')"
+ )
new = url.copy_with(scheme="http")
assert new == URL("http://example.org:123/path/to/somewhere?abc=123#anchor")