Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def merge_headers(
self, headers: HeaderTypes = None
) -> typing.Optional[HeaderTypes]:
"""
Merge a headers argument together with any headers on the client,
to create the headers used for the outgoing request.
"""
if headers or self.headers:
merged_headers = Headers(self.headers)
merged_headers.update(headers)
return merged_headers
return headers
def redirect_headers(self, request: Request, url: URL, method: str) -> Headers:
"""
Return the headers that should be used for the redirect request.
"""
headers = Headers(request.headers)
if url.origin != request.url.origin:
# Strip Authorization headers when responses are redirected away from
# the origin.
headers.pop("Authorization", None)
headers["Host"] = url.authority
if method != request.method and method == "GET":
# If we've switch to a 'GET' request, then strip any headers which
# are only relevant to the request body.
headers.pop("Content-Length", None)
headers.pop("Transfer-Encoding", None)
# We should use the client cookie store to determine any cookie header,
# rather than whatever was on the original outgoing request.
headers.pop("Cookie", None)
def __eq__(self, other: typing.Any) -> bool:
if not isinstance(other, Headers):
return False
return sorted(self._list) == sorted(other._list)
backend=backend,
trust_env=trust_env,
uds=uds,
)
if base_url is None:
self.base_url = URL("", allow_relative=True)
else:
self.base_url = URL(base_url)
if params is None:
params = {}
self.auth = auth
self._params = QueryParams(params)
self._headers = Headers(headers)
self._cookies = Cookies(cookies)
self.timeout = Timeout(timeout)
self.max_redirects = max_redirects
self.trust_env = trust_env
self.dispatch = dispatch
self.netrc = NetRCInfo()
if proxies is None and trust_env:
proxies = typing.cast(ProxiesTypes, get_environment_proxies())
self.proxies: typing.Dict[str, Dispatcher] = _proxies_to_dispatchers(
proxies,
verify=verify,
cert=cert,
http2=http2,
pool_limits=pool_limits,
def redirect_headers(self, request: AsyncRequest, url: URL, method: str) -> Headers:
"""
Return the headers that should be used for the redirect request.
"""
headers = Headers(request.headers)
if url.origin != request.url.origin:
# Strip Authorization headers when responses are redirected away from
# the origin.
headers.pop("Authorization", None)
headers["Host"] = url.authority
if method != request.method and method == "GET":
# If we've switch to a 'GET' request, then strip any headers which
# are only relevant to the request body.
headers.pop("Content-Length", None)
headers.pop("Transfer-Encoding", None)
# We should use the client cookie store to determine any cookie header,
# rather than whatever was on the original outgoing request.
headers.pop("Cookie", None)
def __init__(
self,
status_code: int,
*,
request: Request,
http_version: str = None,
headers: HeaderTypes = None,
stream: ContentStream = None,
content: bytes = None,
history: typing.List["Response"] = None,
elapsed: datetime.timedelta = None,
):
self.status_code = status_code
self.http_version = http_version
self.headers = Headers(headers)
self.request = request
self.elapsed = datetime.timedelta(0) if elapsed is None else elapsed
self.call_next: typing.Optional[typing.Callable] = None
self.history = [] if history is None else list(history)
if stream is None:
self.is_closed = True
self.is_stream_consumed = True
self._raw_content = content or b""
else:
self.is_closed = False
self.is_stream_consumed = False
self._raw_stream = stream
def __init__(
self,
method: str,
url: typing.Union[str, URL],
*,
params: QueryParamTypes = None,
headers: HeaderTypes = None,
cookies: CookieTypes = None,
data: RequestData = None,
files: RequestFiles = None,
json: typing.Any = None,
stream: ContentStream = None,
):
self.method = method.upper()
self.url = URL(url, params=params)
self.headers = Headers(headers)
if cookies:
self._cookies = Cookies(cookies)
self._cookies.set_cookie_header(self)
if stream is not None:
self.stream = stream
else:
self.stream = encode(data, files, json)
self.prepare()