Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import typing
from ..interfaces import Adapter
from ..models import Request, Response
class CookieAdapter(Adapter):
def __init__(self, dispatch: Adapter):
self.dispatch = dispatch
def prepare_request(self, request: Request) -> None:
self.dispatch.prepare_request(request)
async def send(self, request: Request, **options: typing.Any) -> Response:
return await self.dispatch.send(request, **options)
async def close(self) -> None:
await self.dispatch.close()
import typing
from ..config import DEFAULT_MAX_REDIRECTS
from ..constants import codes
from ..exceptions import RedirectBodyUnavailable, RedirectLoop, TooManyRedirects
from ..interfaces import Adapter
from ..models import URL, Headers, Request, Response
class RedirectAdapter(Adapter):
def __init__(self, dispatch: Adapter, max_redirects: int = DEFAULT_MAX_REDIRECTS):
self.dispatch = dispatch
self.max_redirects = max_redirects
def prepare_request(self, request: Request) -> None:
self.dispatch.prepare_request(request)
async def send(self, request: Request, **options: typing.Any) -> Response:
allow_redirects = options.pop("allow_redirects", True) # type: bool
# The following will not typically be specified by the end-user developer,
# but are included in `response.next()` calls.
history = options.pop("history", []) # type: typing.List[Response]
seen_urls = options.pop("seen_urls", set()) # type: typing.Set[URL]
while True:
import typing
from ..interfaces import Adapter
from ..models import Request, Response
class EnvironmentAdapter(Adapter):
def __init__(self, dispatch: Adapter, trust_env: bool = True):
self.dispatch = dispatch
self.trust_env = trust_env
def prepare_request(self, request: Request) -> None:
self.dispatch.prepare_request(request)
async def send(self, request: Request, **options: typing.Any) -> Response:
if self.trust_env:
self.merge_environment_options(options)
return await self.dispatch.send(request, **options)
async def close(self) -> None:
await self.dispatch.close()
def merge_environment_options(self, options: dict) -> None: