from .dispatch.connection_pool import ConnectionPool
from .exceptions import (
ConnectTimeout,
+ CookieConflict,
DecodingError,
InvalidURL,
PoolTimeout,
ProtocolError,
ReadTimeout,
- WriteTimeout,
RedirectBodyUnavailable,
RedirectLoop,
ResponseClosed,
StreamConsumed,
Timeout,
TooManyRedirects,
+ WriteTimeout,
)
from .interfaces import BaseReader, BaseWriter, ConcurrencyBackend, Dispatcher, Protocol
-from .models import URL, Headers, Origin, QueryParams, Request, Response
+from .models import URL, Cookies, Headers, Origin, QueryParams, Request, Response
from .status_codes import codes
__version__ = "0.3.0"
from .models import (
URL,
AuthTypes,
+ Cookies,
+ CookieTypes,
Headers,
HeaderTypes,
QueryParamTypes,
def __init__(
self,
auth: AuthTypes = None,
+ cookies: CookieTypes = None,
ssl: SSLConfig = DEFAULT_SSL_CONFIG,
timeout: TimeoutConfig = DEFAULT_TIMEOUT_CONFIG,
pool_limits: PoolLimits = DEFAULT_POOL_LIMITS,
)
self.auth = auth
+ self.cookies = Cookies(cookies)
self.max_redirects = max_redirects
self.dispatch = dispatch
*,
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
url,
query_params=query_params,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
*,
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
url,
query_params=query_params,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
*,
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = False, # Note: Differs to usual default.
url,
query_params=query_params,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
data=data,
query_params=query_params,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
data=data,
query_params=query_params,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
data=data,
query_params=query_params,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
data=data,
query_params=query_params,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
timeout: TimeoutConfig = None,
) -> Response:
request = Request(
- method, url, data=data, query_params=query_params, headers=headers
+ method,
+ url,
+ data=data,
+ query_params=query_params,
+ headers=headers,
+ cookies=self.merge_cookies(cookies),
)
self.prepare_request(request)
response = await self.send(
def prepare_request(self, request: Request) -> None:
request.prepare()
+ def merge_cookies(self, cookies: CookieTypes = None) -> typing.Optional[CookieTypes]:
+ if cookies or self.cookies:
+ merged_cookies = Cookies(self.cookies)
+ merged_cookies.update(cookies)
+ return merged_cookies
+ return cookies
+
async def send(
self,
request: Request,
request, stream=stream, ssl=ssl, timeout=timeout
)
response.history = list(history)
+ self.cookies.extract_cookies(response)
history = [response] + history
if not response.is_redirect:
break
url = self.redirect_url(request, response)
headers = self.redirect_headers(request, url)
content = self.redirect_content(request, method)
- return Request(method=method, url=url, headers=headers, data=content)
+ cookies = self.merge_cookies(request.cookies)
+ return Request(method=method, url=url, headers=headers, data=content, cookies=cookies)
def redirect_method(self, request: Request, response: Response) -> str:
"""
)
self._loop = asyncio.new_event_loop()
+ @property
+ def cookies(self) -> Cookies:
+ return self._client.cookies
+
def request(
self,
method: str,
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
timeout: TimeoutConfig = None,
) -> SyncResponse:
request = Request(
- method, url, data=data, query_params=query_params, headers=headers
+ method,
+ url,
+ data=data,
+ query_params=query_params,
+ headers=headers,
+ cookies=self._client.merge_cookies(cookies),
)
self.prepare_request(request)
response = self.send(
*,
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
"GET",
url,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
*,
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
"OPTIONS",
url,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
*,
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = False, # Note: Differs to usual default.
"HEAD",
url,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
url,
data=data,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
url,
data=data,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
url,
data=data,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
stream: bool = False,
auth: AuthTypes = None,
allow_redirects: bool = True,
url,
data=data,
headers=headers,
+ cookies=cookies,
stream=stream,
auth=auth,
allow_redirects=allow_redirects,
"""
URL was missing a hostname, or was not one of HTTP/HTTPS.
"""
+
+
+class CookieConflict(Exception):
+ """
+ Attempted to lookup a cookie by name, but multiple cookies existed.
+ """
import asyncio
import cgi
+import email.message
import typing
+import urllib.request
+from collections.abc import MutableMapping
+from http.cookiejar import Cookie, CookieJar
from urllib.parse import parse_qsl, urlencode
import chardet
MultiDecoder,
)
from .exceptions import (
+ CookieConflict,
HttpError,
InvalidURL,
ResponseClosed,
typing.List[typing.Tuple[typing.AnyStr, typing.AnyStr]],
]
+CookieTypes = typing.Union["Cookies", CookieJar, typing.Dict[str, str]]
+
AuthTypes = typing.Union[
typing.Tuple[typing.Union[str, bytes], typing.Union[str, bytes]],
typing.Callable[["Request"], "Request"],
data: RequestData = b"",
query_params: QueryParamTypes = None,
headers: HeaderTypes = None,
+ cookies: CookieTypes = None,
):
self.method = method.upper()
self.url = URL(url, query_params=query_params)
self.headers = Headers(headers)
+ if cookies:
+ self._cookies = Cookies(cookies)
+ self._cookies.set_cookie_header(self)
if isinstance(data, bytes):
self.is_streaming = False
for item in reversed(auto_headers):
self.headers.raw.insert(0, item)
+ @property
+ def cookies(self) -> "Cookies":
+ if not hasattr(self, "_cookies"):
+ self._cookies = Cookies()
+ return self._cookies
+
def __repr__(self) -> str:
class_name = self.__class__.__name__
url = str(self.url)
if message:
raise HttpError(message)
+ @property
+ def cookies(self) -> "Cookies":
+ if not hasattr(self, "_cookies"):
+ assert self.request is not None
+ self._cookies = Cookies()
+ self._cookies.extract_cookies(self)
+ return self._cookies
+
def __repr__(self) -> str:
return f"<Response({self.status_code}, {self.reason_phrase!r})>"
def close(self) -> None:
return self._loop.run_until_complete(self._response.close())
+ @property
+ def cookies(self) -> "Cookies":
+ return self._response.cookies
+
def __repr__(self) -> str:
return f"<Response({self.status_code}, {self.reason_phrase!r})>"
+
+
+class Cookies(MutableMapping):
+ """
+ HTTP Cookies, as a mutable mapping.
+ """
+
+ def __init__(self, cookies: CookieTypes = None) -> None:
+ if cookies is None or isinstance(cookies, dict):
+ self.jar = CookieJar()
+ if isinstance(cookies, dict):
+ for key, value in cookies.items():
+ self.set(key, value)
+ elif isinstance(cookies, Cookies):
+ self.jar = CookieJar()
+ for cookie in cookies.jar:
+ self.jar.set_cookie(cookie)
+ else:
+ self.jar = cookies
+
+ def extract_cookies(self, response: Response) -> None:
+ """
+ Loads any cookies based on the response `Set-Cookie` headers.
+ """
+ assert response.request is not None
+ urlib_response = self._CookieCompatResponse(response)
+ urllib_request = self._CookieCompatRequest(response.request)
+
+ self.jar.extract_cookies(urlib_response, urllib_request) # type: ignore
+
+ def set_cookie_header(self, request: Request) -> None:
+ """
+ Sets an appropriate 'Cookie:' HTTP header on the `Request`.
+ """
+ urllib_request = self._CookieCompatRequest(request)
+ self.jar.add_cookie_header(urllib_request)
+
+ def set(self, name: str, value: str, domain: str = "", path: str = "/") -> None:
+ """
+ Set a cookie value by name. May optionally include domain and path.
+ """
+ kwargs = dict(
+ version=0,
+ name=name,
+ value=value,
+ port=None,
+ port_specified=False,
+ domain=domain,
+ domain_specified=bool(domain),
+ domain_initial_dot=domain.startswith("."),
+ path=path,
+ path_specified=bool(path),
+ secure=False,
+ expires=None,
+ discard=True,
+ comment=None,
+ comment_url=None,
+ rest={"HttpOnly": None},
+ rfc2109=False,
+ )
+ cookie = Cookie(**kwargs) # type: ignore
+ self.jar.set_cookie(cookie)
+
+ def get( # type: ignore
+ self, name: str, default: str = None, domain: str = None, path: str = None
+ ) -> typing.Optional[str]:
+ """
+ Get a cookie by name. May optionally include domain and path
+ in order to specify exactly which cookie to retrieve.
+ """
+ value = None
+ for cookie in self.jar:
+ if cookie.name == name:
+ if domain is None or cookie.domain == domain: # type: ignore
+ if path is None or cookie.path == path:
+ if value is not None:
+ message = f"Multiple cookies exist with name={name}"
+ raise CookieConflict(message)
+ value = cookie.value
+
+ if value is None:
+ return default
+ return value
+
+ def delete(self, name: str, domain: str = None, path: str = None) -> None:
+ """
+ Delete a cookie by name. May optionally include domain and path
+ in order to specify exactly which cookie to delete.
+ """
+ if domain is not None and path is not None:
+ return self.jar.clear(domain, path, name)
+
+ remove = []
+ for cookie in self.jar:
+ if cookie.name == name:
+ if domain is None or cookie.domain == domain: # type: ignore
+ if path is None or cookie.path == path:
+ remove.append(cookie)
+
+ for cookie in remove:
+ self.jar.clear(cookie.domain, cookie.path, cookie.name) # type: ignore
+
+ def clear(self, domain: str = None, path: str = None) -> None:
+ """
+ Delete all cookies. Optionally include a domain and path in
+ order to only delete a subset of all the cookies.
+ """
+ args = []
+ if domain is not None:
+ args.append(domain)
+ if path is not None:
+ assert domain is not None
+ args.append(path)
+ self.jar.clear(*args)
+
+ def update(self, cookies: CookieTypes) -> None: # type: ignore
+ cookies = Cookies(cookies)
+ for cookie in cookies.jar:
+ self.jar.set_cookie(cookie)
+
+ def __setitem__(self, name: str, value: str) -> None:
+ return self.set(name, value)
+
+ def __getitem__(self, name: str) -> str:
+ value = self.get(name)
+ if value is None:
+ raise KeyError(name)
+ return value
+
+ def __delitem__(self, name: str) -> None:
+ return self.delete(name)
+
+ def __len__(self) -> int:
+ return len(self.jar)
+
+ def __iter__(self) -> typing.Iterator[str]:
+ return (cookie.name for cookie in self.jar)
+
+ def __bool__(self) -> bool:
+ for cookie in self.jar:
+ return True
+ return False
+
+ class _CookieCompatRequest(urllib.request.Request):
+ """
+ Wraps a `Request` instance up in a compatability interface suitable
+ for use with `CookieJar` operations.
+ """
+
+ def __init__(self, request: Request) -> None:
+ super().__init__(
+ url=str(request.url),
+ headers=dict(request.headers),
+ method=request.method,
+ )
+ self.request = request
+
+ def add_unredirected_header(self, key: str, value: str) -> None:
+ super().add_unredirected_header(key, value)
+ self.request.headers[key] = value
+
+ class _CookieCompatResponse:
+ """
+ Wraps a `Request` instance up in a compatability interface suitable
+ for use with `CookieJar` operations.
+ """
+
+ def __init__(self, response: Response):
+ self.response = response
+
+ def info(self) -> email.message.Message:
+ info = email.message.Message()
+ for key, value in self.response.headers.items():
+ info[key] = value
+ return info
set -x
-PYTHONPATH=. ${PREFIX}pytest --ignore venv --cov tests --cov ${PACKAGE} --cov-report= --cov-fail-under=100 ${@}
-${PREFIX}coverage report -m
+PYTHONPATH=. ${PREFIX}pytest --ignore venv --cov tests --cov ${PACKAGE} --cov-report= ${@}
+${PREFIX}coverage report --show-missing --fail-under=100
--- /dev/null
+import json
+from http.cookiejar import Cookie, CookieJar
+
+import pytest
+
+from httpcore import (
+ URL,
+ Client,
+ Cookies,
+ Dispatcher,
+ Request,
+ Response,
+ SSLConfig,
+ TimeoutConfig,
+)
+
+
+class MockDispatch(Dispatcher):
+ async def send(
+ self,
+ request: Request,
+ stream: bool = False,
+ ssl: SSLConfig = None,
+ timeout: TimeoutConfig = None,
+ ) -> Response:
+ if request.url.path.startswith("/echo_cookies"):
+ body = json.dumps({"cookies": request.headers.get("Cookie")}).encode()
+ return Response(200, content=body, request=request)
+ elif request.url.path.startswith("/set_cookie"):
+ headers = {"set-cookie": "example-name=example-value"}
+ return Response(200, headers=headers, request=request)
+
+
+def test_set_cookie():
+ """
+ Send a request including a cookie.
+ """
+ url = "http://example.org/echo_cookies"
+ cookies = {"example-name": "example-value"}
+
+ with Client(dispatch=MockDispatch()) as client:
+ response = client.get(url, cookies=cookies)
+
+ assert response.status_code == 200
+ assert json.loads(response.text) == {"cookies": "example-name=example-value"}
+
+
+def test_set_cookie_with_cookiejar():
+ """
+ Send a request including a cookie, using a `CookieJar` instance.
+ """
+
+ url = "http://example.org/echo_cookies"
+ cookies = CookieJar()
+ cookie = Cookie(
+ version=0,
+ name="example-name",
+ value="example-value",
+ port=None,
+ port_specified=False,
+ domain="",
+ domain_specified=False,
+ domain_initial_dot=False,
+ path="/",
+ path_specified=True,
+ secure=False,
+ expires=None,
+ discard=True,
+ comment=None,
+ comment_url=None,
+ rest={"HttpOnly": None},
+ rfc2109=False,
+ )
+ cookies.set_cookie(cookie)
+
+ with Client(dispatch=MockDispatch()) as client:
+ response = client.get(url, cookies=cookies)
+
+ assert response.status_code == 200
+ assert json.loads(response.text) == {"cookies": "example-name=example-value"}
+
+
+def test_set_cookie_with_cookies_model():
+ """
+ Send a request including a cookie, using a `Cookies` instance.
+ """
+
+ url = "http://example.org/echo_cookies"
+ cookies = Cookies()
+ cookies["example-name"] = "example-value"
+
+ with Client(dispatch=MockDispatch()) as client:
+ response = client.get(url, cookies=cookies)
+
+ assert response.status_code == 200
+ assert json.loads(response.text) == {"cookies": "example-name=example-value"}
+
+
+def test_get_cookie():
+ url = "http://example.org/set_cookie"
+
+ with Client(dispatch=MockDispatch()) as client:
+ response = client.get(url)
+
+ assert response.status_code == 200
+ assert response.cookies["example-name"] == "example-value"
+ assert client.cookies["example-name"] == "example-value"
+
+
+def test_cookie_persistence():
+ """
+ Ensure that Client instances persist cookies between requests.
+ """
+ with Client(dispatch=MockDispatch()) as client:
+ response = client.get("http://example.org/echo_cookies")
+ assert response.status_code == 200
+ assert json.loads(response.text) == {"cookies": None}
+
+ response = client.get("http://example.org/set_cookie")
+ assert response.status_code == 200
+ assert response.cookies["example-name"] == "example-value"
+ assert client.cookies["example-name"] == "example-value"
+
+ response = client.get("http://example.org/echo_cookies")
+ assert response.status_code == 200
+ assert json.loads(response.text) == {"cookies": "example-name=example-value"}
--- /dev/null
+import pytest
+
+from httpcore import CookieConflict, Cookies
+
+
+def test_cookies():
+ cookies = Cookies({"name": "value"})
+ assert cookies["name"] == "value"
+ assert "name" in cookies
+ assert len(cookies) == 1
+ assert dict(cookies) == {"name": "value"}
+ assert bool(cookies) is True
+
+ del cookies["name"]
+ assert "name" not in cookies
+ assert len(cookies) == 0
+ assert dict(cookies) == {}
+ assert bool(cookies) is False
+
+
+def test_cookies_update():
+ cookies = Cookies()
+ more_cookies = Cookies()
+ more_cookies.set("name", "value", domain="example.com")
+
+ cookies.update(more_cookies)
+ assert dict(cookies) == {"name": "value"}
+ assert cookies.get("name", domain="example.com") == "value"
+
+
+def test_cookies_with_domain():
+ cookies = Cookies()
+ cookies.set("name", "value", domain="example.com")
+ cookies.set("name", "value", domain="example.org")
+
+ with pytest.raises(CookieConflict):
+ cookies["name"]
+
+ cookies.clear(domain="example.com")
+ assert len(cookies) == 1
+
+
+def test_cookies_with_domain_and_path():
+ cookies = Cookies()
+ cookies.set("name", "value", domain="example.com", path="/subpath/1")
+ cookies.set("name", "value", domain="example.com", path="/subpath/2")
+ cookies.clear(domain="example.com", path="/subpath/1")
+ assert len(cookies) == 1
+ cookies.delete("name", domain="example.com", path="/subpath/2")
+ assert len(cookies) == 0
PoolLimits,
PoolTimeout,
ReadTimeout,
- WriteTimeout,
TimeoutConfig,
+ WriteTimeout,
)
projects / thirdparty / httpx.git / commitdiff
