```
"""
with Client(
- proxies=proxies, cert=cert, verify=verify, timeout=timeout, trust_env=trust_env,
+ proxies=proxies, cert=cert, verify=verify, timeout=timeout, trust_env=trust_env
) as client:
return client.request(
method=method,
return url.copy_with(path=url.path + "/")
def _get_proxy_map(
- self, proxies: typing.Optional[ProxiesTypes], allow_env_proxies: bool,
+ self, proxies: typing.Optional[ProxiesTypes], allow_env_proxies: bool
) -> typing.Dict[str, typing.Optional[Proxy]]:
if proxies is None:
if allow_env_proxies:
cookies=cookies,
)
return self.send(
- request, auth=auth, allow_redirects=allow_redirects, timeout=timeout,
+ request, auth=auth, allow_redirects=allow_redirects, timeout=timeout
)
def send(
auth = self._build_request_auth(request, auth)
response = self._send_handling_redirects(
- request, auth=auth, timeout=timeout, allow_redirects=allow_redirects,
+ request, auth=auth, timeout=timeout, allow_redirects=allow_redirects
)
if not stream:
cookies=cookies,
)
response = await self.send(
- request, auth=auth, allow_redirects=allow_redirects, timeout=timeout,
+ request, auth=auth, allow_redirects=allow_redirects, timeout=timeout
)
return response
auth = self._build_request_auth(request, auth)
response = await self._send_handling_redirects(
- request, auth=auth, timeout=timeout, allow_redirects=allow_redirects,
+ request, auth=auth, timeout=timeout, allow_redirects=allow_redirects
)
if not stream:
history.append(response)
async def _send_single_request(
- self, request: Request, timeout: Timeout,
+ self, request: Request, timeout: Timeout
) -> Response:
"""
Sends a single request, without handling any redirections.
class Proxy:
def __init__(
- self, url: URLTypes, *, headers: HeaderTypes = None, mode: str = "DEFAULT",
+ self, url: URLTypes, *, headers: HeaderTypes = None, mode: str = "DEFAULT"
):
url = URL(url)
headers = Headers(headers)
# Tests & Linting
autoflake
-black
+black==20.8b1
cryptography
flake8
flake8-bugbear
return b"HTTP/1.1", 200, b"", [], body
def challenge_send(
- self, method: bytes, headers: typing.List[typing.Tuple[bytes, bytes]],
+ self, method: bytes, headers: typing.List[typing.Tuple[bytes, bytes]]
) -> typing.Tuple[
bytes, int, bytes, typing.List[typing.Tuple[bytes, bytes]], ContentStream
]:
"Content-Type": "text-plain; charset=utf-8"
} # Deliberately incorrect charset
response = httpx.Response(
- 200, content="Latin 1: ÿ".encode("latin-1"), headers=headers, request=REQUEST,
+ 200, content="Latin 1: ÿ".encode("latin-1"), headers=headers, request=REQUEST
)
response.encoding = "latin-1"
assert response.text == "Latin 1: ÿ"
@pytest.mark.parametrize(
["pattern", "url", "expected"],
[
- ("http://example.com", "http://example.com", True,),
- ("http://example.com", "https://example.com", False,),
- ("http://example.com", "http://other.com", False,),
- ("http://example.com:123", "http://example.com:123", True,),
- ("http://example.com:123", "http://example.com:456", False,),
- ("http://example.com:123", "http://example.com", False,),
- ("all://example.com", "http://example.com", True,),
- ("all://example.com", "https://example.com", True,),
- ("http://", "http://example.com", True,),
- ("http://", "https://example.com", False,),
- ("all://", "https://example.com:123", True,),
- ("", "https://example.com:123", True,),
+ ("http://example.com", "http://example.com", True),
+ ("http://example.com", "https://example.com", False),
+ ("http://example.com", "http://other.com", False),
+ ("http://example.com:123", "http://example.com:123", True),
+ ("http://example.com:123", "http://example.com:456", False),
+ ("http://example.com:123", "http://example.com", False),
+ ("all://example.com", "http://example.com", True),
+ ("all://example.com", "https://example.com", True),
+ ("http://", "http://example.com", True),
+ ("http://", "https://example.com", False),
+ ("all://", "https://example.com:123", True),
+ ("", "https://example.com:123", True),
],
)
def test_url_matches(pattern, url, expected):