!!! hint
A Client instance is equivalent to a Session instance in `requests`.
+!!! note
+ Starting from `httpx==0.10.0`, the default and recommended Client class is `AsyncClient`. This should help prevent breaking changes once sync support is reintroduced.
+
+ A `Client` synonym remains for compatibility with `httpx==0.9.*` releases, but you are encouraged to migrate to `AsyncClient` as soon as possible.
+
### Usage
The recommended way to use a `Client` is as a context manager. This will ensure that connections are properly cleaned up when leaving the `with` block:
```python
->>> async with httpx.Client() as client:
+>>> async with httpx.AsyncClient() as client:
... r = await client.get('https://example.com')
...
>>> r
Alternatively, you can explicitly close the connection pool without block-usage using `.aclose()`:
```python
->>> client = httpx.Client()
+>>> client = httpx.AsyncClient()
>>> try:
... r = await client.get('https://example.com')
... finally:
```python
>>> url = 'http://httpbin.org/headers'
>>> headers = {'user-agent': 'my-app/0.0.1'}
->>> async with httpx.Client(headers=headers) as client:
+>>> async with httpx.AsyncClient(headers=headers) as client:
... r = await client.get(url)
...
>>> r.json()['headers']['User-Agent']
- For headers, query parameters and cookies, the values are merged into one.
- For all other parameters, the request-level value is used.
-Additionally, `Client` constructor accepts some parameters that aren't available at the request level.
+Additionally, `Client` accepts some parameters that aren't available at the request level.
One particularly useful parameter is `base_url`, which allows you to define a base URL to prepend to all outgoing requests:
```python
->>> async with httpx.Client(base_url='http://httpbin.org') as client:
+>>> async with httpx.AsyncClient(base_url='http://httpbin.org') as client:
... r = await client.get('/headers')
...
>>> r.request.url
URL('http://httpbin.org/headers')
```
-For a list of all available client-level parameters, see the [`Client` API reference](/api/#client).
+For a list of all available client-level parameters, see the [`AsyncClient` API reference](/api/#async-client).
## Calling into Python Web Apps
```python
>>> import httpx
->>> async with httpx.Client(app=app) as client:
+>>> async with httpx.AsyncClient(app=app) as client:
... r = client.get('http://example/')
... assert r.status_code == 200
... assert r.text == "Hello World!"
# Instantiate a client that makes ASGI requests with a client IP of "1.2.3.4",
# on port 123.
dispatch = httpx.dispatch.ASGIDispatch(app=app, client=("1.2.3.4", 123))
-async with httpx.Client(dispatch=dispatch) as client:
+async with httpx.AsyncClient(dispatch=dispatch) as client:
...
```
make modifications before sending the request.
```python
->>> async with httpx.Client() as client:
+>>> async with httpx.AsyncClient() as client:
... req = client.build_request("OPTIONS", "https://example.com")
... req.url.full_path = "*" # Build an 'OPTIONS *' request for CORS
... r = await client.send(req)
When using `Client` instances, `trust_env` should be set on the client itself, rather that on the request methods:
```python
-client = httpx.Client(trust_env=False)
+client = httpx.AsyncClient(trust_env=False)
```
## Unix Domain Sockets
import httpx
-async with httpx.Client(uds="/var/run/docker.sock") as client:
+async with httpx.AsyncClient(uds="/var/run/docker.sock") as client:
# This request will connect through the socket file.
resp = await client.get("http://localhost/version")
```
... "http": "http://127.0.0.1:3080",
... "https": "http://127.0.0.1:3081"
... }
->>> async with httpx.Client(proxies=proxies) as client:
+>>> async with httpx.AsyncClient(proxies=proxies) as client:
... ...
```
... "http": "...", # Scheme
... "all": "...", # All
... }
->>> async with httpx.Client(proxies=proxies) as client:
+>>> async with httpx.AsyncClient(proxies=proxies) as client:
... ...
...
>>> proxy = "..." # Shortcut for {'all': '...'}
->>> async with httpx.Client(proxies=proxy) as client:
+>>> async with httpx.AsyncClient(proxies=proxy) as client:
... ...
```
proxy_url="https://127.0.0.1",
proxy_mode="TUNNEL_ONLY" # May be "TUNNEL_ONLY" or "FORWARD_ONLY". Defaults to "DEFAULT".
)
-async with httpx.Client(proxies=proxy) as client:
+async with httpx.AsyncClient(proxies=proxy) as client:
# This request will be tunneled instead of forwarded.
r = await client.get("http://example.com")
```
await httpx.get('http://example.com/api/v1/example', timeout=10.0)
# Using a client instance:
-async with httpx.Client() as client:
+async with httpx.AsyncClient() as client:
await client.get("http://example.com/api/v1/example", timeout=10.0)
```
await httpx.get('http://example.com/api/v1/example', timeout=None)
# Using a client instance:
-async with httpx.Client() as client:
+async with httpx.AsyncClient() as client:
await client.get("http://example.com/api/v1/example", timeout=None)
```
`timeout` being used as the default for requests made with this client:
```python
-client = httpx.Client() # Use a default 5s timeout everywhere.
-client = httpx.Client(timeout=10.0) # Use a default 10s timeout everywhere.
-client = httpx.Client(timeout=None) # Disable all timeouts by default.
+client = httpx.AsyncClient() # Use a default 5s timeout everywhere.
+client = httpx.AsyncClient(timeout=10.0) # Use a default 10s timeout everywhere.
+client = httpx.AsyncClient(timeout=None) # Disable all timeouts by default.
```
### Fine tuning the configuration
```python
# A client with a 60s timeout for connecting, and a 10s timeout elsewhere.
timeout = httpx.Timeout(10.0, connect_timeout=60.0)
-client = httpx.Client(timeout=timeout)
+client = httpx.AsyncClient(timeout=timeout)
response = await client.get('http://example.com/')
```
If you're using a `Client()` instance, then you should pass any SSL settings when instantiating the client.
```python
-client = httpx.Client(verify=False)
+client = httpx.AsyncClient(verify=False)
```
The `client.get(...)` method and other request methods *do not* support changing the SSL settings on a per-request basis. If you need different SSL settings in different cases you should use more that one client instance, with different settings on each. Each client will then be using an isolated connection pool with a specific fixed SSL configuration on all connections within that pool.
`backend` argument...
```python
-client = httpx.Client(backend='auto') # Autodetection. The default case.
-client = httpx.Client(backend='asyncio') # Use asyncio as the backend.
-client = httpx.Client(backend='trio') # Use trio as the backend.
+client = httpx.AsyncClient(backend='auto') # Autodetection. The default case.
+client = httpx.AsyncClient(backend='asyncio') # Use asyncio as the backend.
+client = httpx.AsyncClient(backend='trio') # Use trio as the backend.
```
### [AsyncIO](https://docs.python.org/3/library/asyncio.html)
import httpx
async def main():
- client = httpx.Client()
+ client = httpx.AsyncClient()
response = await client.get('https://www.example.com/')
print(response)
import trio
async def main():
- client = httpx.Client()
+ client = httpx.AsyncClient()
response = await client.get('https://www.example.com/')
print(response)
::: httpx.delete
:docstring:
-## `Client`
+## `AsyncClient`
-::: httpx.Client
+::: httpx.AsyncClient
:docstring:
:members: headers cookies params request get head options post put patch delete build_request send aclose
## Client instances
-The HTTPX equivalent of `requests.Session` is `httpx.Client`.
+The HTTPX equivalent of `requests.Session` is `httpx.AsyncClient`.
```python
session = requests.Session(**kwargs)
is generally equivalent to
```python
-client = httpx.Client(**kwargs)
+client = httpx.AsyncClient(**kwargs)
```
## Mocking
The HTTPX library can be configured via environment variables.
Environment variables are used by default. To ignore environment variables, `trust_env` has to be set `False`. There are two ways to set `trust_env` to disable environment variables:
-* On the client via `httpx.Client(trust_env=False)`.
+* On the client via `httpx.AsyncClient(trust_env=False)`.
* Using the top-level API, such as `httpx.get("<url>", trust_env=False)`.
Here is a list of environment variables that HTTPX recognizes and what function they serve:
# test_script.py
import httpx
-with httpx.Client() as client:
+with httpx.AsyncClient() as client:
r = client.get("https://google.com")
```
# test_script.py
import httpx
-with httpx.Client() as client:
+with httpx.AsyncClient() as client:
r = client.get("https://google.com")
```
HTTP/2 support enabled:
```python
-client = httpx.Client(http2=True)
+client = httpx.AsyncClient(http2=True)
...
```
is exited.
```python
-async with httpx.Client(http2=True) as client:
+async with httpx.AsyncClient(http2=True) as client:
...
```
the `.http_version` property on the response.
```python
-client = httpx.Client(http2=True)
+client = httpx.AsyncClient(http2=True)
response = await client.get(...)
print(response.http_version) # "HTTP/1.0", "HTTP/1.1", or "HTTP/2".
```
from .__version__ import __description__, __title__, __version__
from .api import delete, get, head, options, patch, post, put, request, stream
from .auth import BasicAuth, DigestAuth
-from .client import Client
+from .client import AsyncClient, Client
from .config import TimeoutConfig # For 0.8 backwards compat.
from .config import PoolLimits, Timeout
from .dispatch.proxy_http import HTTPProxy, HTTPProxyMode
"request",
"stream",
"codes",
+ "AsyncClient",
"BasicAuth",
"Client",
"DigestAuth",
import typing
from .auth import AuthTypes
-from .client import Client, StreamContextManager
+from .client import AsyncClient, StreamContextManager
from .config import DEFAULT_TIMEOUT_CONFIG, CertTypes, TimeoutTypes, VerifyTypes
from .models import (
CookieTypes,
<Response [200 OK]>
```
"""
- async with Client(
+ async with AsyncClient(
cert=cert, verify=verify, timeout=timeout, trust_env=trust_env,
) as client:
return await client.request(
cert: CertTypes = None,
trust_env: bool = True,
) -> StreamContextManager:
- client = Client(cert=cert, verify=verify, trust_env=trust_env)
+ client = AsyncClient(cert=cert, verify=verify, trust_env=trust_env)
request = Request(
method=method,
url=url,
logger = get_logger(__name__)
-class Client:
+class AsyncClient:
"""
- An HTTP client, with connection pooling, HTTP/2, redirects, cookie persistence, etc.
+ An asynchronous HTTP client, with connection pooling, HTTP/2, redirects,
+ cookie persistence, etc.
Usage:
- ```
- >>> client = httpx.Client()
+ ```python
+ >>> client = httpx.AsyncClient()
>>> response = client.get('https://example.org')
```
async def aclose(self) -> None:
await self.dispatch.close()
- async def __aenter__(self) -> "Client":
+ async def __aenter__(self) -> "AsyncClient":
return self
async def __aexit__(
class StreamContextManager:
def __init__(
self,
- client: Client,
+ client: AsyncClient,
request: Request,
*,
auth: AuthTypes = None,
await self.response.aclose()
if self.close_client:
await self.client.aclose()
+
+
+# For compatibility with 0.9.x.
+Client = AsyncClient
and will setup an appropriate dispatch class:
```
- client = httpx.Client(app=app)
+ client = httpx.AsyncClient(app=app)
```
Alternatively, you can setup the dispatch instance explicitly.
root_path="/submount",
client=("1.2.3.4", 123)
)
- client = httpx.Client(dispatch=dispatch)
+ client = httpx.AsyncClient(dispatch=dispatch)
```
Arguments:
Stubs out the interface, as well as providing a `.request()` convenience
implementation, to make it easy to use or test stand-alone dispatchers,
- without requiring a complete `Client` instance.
+ without requiring a complete `AsyncClient` instance.
"""
async def request(
@pytest.mark.usefixtures("async_environment")
async def test_get(server):
url = server.url
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.get(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
async def test_build_request(server):
url = server.url.copy_with(path="/echo_headers")
headers = {"Custom-header": "value"}
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
request = client.build_request("GET", url)
request.headers.update(headers)
response = await client.send(request)
@pytest.mark.usefixtures("async_environment")
async def test_post(server):
url = server.url
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.post(url, data=b"Hello, world!")
assert response.status_code == 200
@pytest.mark.usefixtures("async_environment")
async def test_post_json(server):
url = server.url
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.post(url, json={"text": "Hello, world!"})
assert response.status_code == 200
@pytest.mark.usefixtures("async_environment")
async def test_stream_response(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
async with client.stream("GET", server.url) as response:
body = await response.aread()
@pytest.mark.usefixtures("async_environment")
async def test_access_content_stream_response(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
async with client.stream("GET", server.url) as response:
pass
yield b"Hello, "
yield b"world!"
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.request("POST", server.url, data=hello_world())
assert response.status_code == 200
@pytest.mark.usefixtures("async_environment")
async def test_raise_for_status(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
for status_code in (200, 400, 404, 500, 505):
response = await client.request(
"GET", server.url.copy_with(path=f"/status/{status_code}")
@pytest.mark.usefixtures("async_environment")
async def test_options(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.options(server.url)
assert response.status_code == 200
assert response.text == "Hello, world!"
@pytest.mark.usefixtures("async_environment")
async def test_head(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.head(server.url)
assert response.status_code == 200
assert response.text == ""
@pytest.mark.usefixtures("async_environment")
async def test_put(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.put(server.url, data=b"Hello, world!")
assert response.status_code == 200
@pytest.mark.usefixtures("async_environment")
async def test_patch(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.patch(server.url, data=b"Hello, world!")
assert response.status_code == 200
@pytest.mark.usefixtures("async_environment")
async def test_delete(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.delete(server.url)
assert response.status_code == 200
assert response.text == "Hello, world!"
headers = {"Expect": "100-continue"}
data = b"Echo request body"
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.post(
server.url.copy_with(path="/echo_body"), headers=headers, data=data
)
url = uds_server.url
uds = uds_server.config.uds
assert uds is not None
- async with httpx.Client(uds=uds) as client:
+ async with httpx.AsyncClient(uds=uds) as client:
response = await client.get(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
async def test_explicit_backend(server, async_environment):
- async with httpx.Client(backend=async_environment) as client:
+ async with httpx.AsyncClient(backend=async_environment) as client:
response = await client.get(server.url)
assert response.status_code == 200
assert response.text == "Hello, world!"
import pytest
-from httpx import URL, Client, DigestAuth, ProtocolError, Request, Response
+from httpx import URL, AsyncClient, DigestAuth, ProtocolError, Request, Response
from httpx.config import CertTypes, TimeoutTypes, VerifyTypes
from httpx.dispatch.base import Dispatcher
url = "https://example.org/"
auth = ("tomchristie", "password123")
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url, auth=auth)
assert response.status_code == 200
async def test_basic_auth_in_url() -> None:
url = "https://tomchristie:password123@example.org/"
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url)
assert response.status_code == 200
url = "https://example.org/"
auth = ("tomchristie", "password123")
- client = Client(dispatch=MockDispatch(), auth=auth)
+ client = AsyncClient(dispatch=MockDispatch(), auth=auth)
response = await client.get(url)
assert response.status_code == 200
request.headers["Authorization"] = "Token 123"
return request
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url, auth=auth)
assert response.status_code == 200
os.environ["NETRC"] = "tests/.netrc"
url = "http://netrcexample.org"
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url)
assert response.status_code == 200
os.environ["NETRC"] = "tests/.netrc"
url = "http://netrcexample.org"
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url, headers={"Authorization": "Override"})
assert response.status_code == 200
os.environ["NETRC"] = "tests/.netrc"
url = "http://netrcexample.org"
- client = Client(dispatch=MockDispatch(), trust_env=False)
+ client = AsyncClient(dispatch=MockDispatch(), trust_env=False)
response = await client.get(url)
assert response.status_code == 200
assert response.json() == {"auth": None}
- client = Client(dispatch=MockDispatch(), trust_env=True)
+ client = AsyncClient(dispatch=MockDispatch(), trust_env=True)
response = await client.get(url)
assert response.status_code == 200
url = "https://example.org/"
auth = ("example-username", "example-password")
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url, auth=auth)
assert "'authorization': '[secure]'" in str(response.request.headers)
@pytest.mark.asyncio
async def test_auth_invalid_type() -> None:
url = "https://example.org/"
- client = Client(
+ client = AsyncClient(
dispatch=MockDispatch(), auth="not a tuple, not a callable", # type: ignore
)
with pytest.raises(TypeError):
url = "https://example.org/"
auth = DigestAuth(username="tomchristie", password="password123")
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url, auth=auth)
assert response.status_code == 200
auth = DigestAuth(username="tomchristie", password="password123")
auth_header = 'Digest realm="realm@host.com",qop="auth",nonce="abc",opaque="xyz"'
- client = Client(dispatch=MockDispatch(auth_header=auth_header, status_code=200))
+ client = AsyncClient(
+ dispatch=MockDispatch(auth_header=auth_header, status_code=200)
+ )
response = await client.get(url, auth=auth)
assert response.status_code == 200
url = "https://example.org/"
auth = DigestAuth(username="tomchristie", password="password123")
- client = Client(dispatch=MockDispatch(auth_header="", status_code=401))
+ client = AsyncClient(dispatch=MockDispatch(auth_header="", status_code=401))
response = await client.get(url, auth=auth)
assert response.status_code == 401
url = "https://example.org/"
auth = DigestAuth(username="tomchristie", password="password123")
- client = Client(dispatch=MockDigestAuthDispatch(algorithm=algorithm))
+ client = AsyncClient(dispatch=MockDigestAuthDispatch(algorithm=algorithm))
response = await client.get(url, auth=auth)
assert response.status_code == 200
url = "https://example.org/"
auth = DigestAuth(username="tomchristie", password="password123")
- client = Client(dispatch=MockDigestAuthDispatch(qop=""))
+ client = AsyncClient(dispatch=MockDigestAuthDispatch(qop=""))
response = await client.get(url, auth=auth)
assert response.status_code == 200
url = "https://example.org/"
auth = DigestAuth(username="tomchristie", password="password123")
- client = Client(dispatch=MockDigestAuthDispatch(qop=qop))
+ client = AsyncClient(dispatch=MockDigestAuthDispatch(qop=qop))
await client.get(url, auth=auth)
async def test_digest_auth_qop_auth_int_not_implemented() -> None:
url = "https://example.org/"
auth = DigestAuth(username="tomchristie", password="password123")
- client = Client(dispatch=MockDigestAuthDispatch(qop="auth-int"))
+ client = AsyncClient(dispatch=MockDigestAuthDispatch(qop="auth-int"))
with pytest.raises(NotImplementedError):
await client.get(url, auth=auth)
async def test_digest_auth_qop_must_be_auth_or_auth_int() -> None:
url = "https://example.org/"
auth = DigestAuth(username="tomchristie", password="password123")
- client = Client(dispatch=MockDigestAuthDispatch(qop="not-auth"))
+ client = AsyncClient(dispatch=MockDigestAuthDispatch(qop="not-auth"))
with pytest.raises(ProtocolError):
await client.get(url, auth=auth)
url = "https://example.org/"
auth = DigestAuth(username="tomchristie", password="password123")
- client = Client(dispatch=MockDigestAuthDispatch(send_response_after_attempt=2))
+ client = AsyncClient(dispatch=MockDigestAuthDispatch(send_response_after_attempt=2))
response = await client.get(url, auth=auth)
assert response.status_code == 401
) -> None:
url = "https://example.org/"
auth = DigestAuth(username="tomchristie", password="password123")
- client = Client(dispatch=MockDispatch(auth_header=auth_header, status_code=401))
+ client = AsyncClient(
+ dispatch=MockDispatch(auth_header=auth_header, status_code=401)
+ )
with pytest.raises(ProtocolError):
await client.get(url, auth=auth)
@pytest.mark.asyncio
async def test_get(server):
url = server.url
- async with httpx.Client() as http:
+ async with httpx.AsyncClient() as http:
response = await http.get(url)
assert response.status_code == 200
assert response.url == url
url = server.url.copy_with(path="/echo_headers")
headers = {"Custom-header": "value"}
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
request = client.build_request("GET", url)
request.headers.update(headers)
response = await client.send(request)
@pytest.mark.asyncio
async def test_post(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.post(server.url, data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@pytest.mark.asyncio
async def test_post_json(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.post(server.url, json={"text": "Hello, world!"})
assert response.status_code == 200
assert response.reason_phrase == "OK"
@pytest.mark.asyncio
async def test_stream_response(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
async with client.stream("GET", server.url) as response:
content = await response.aread()
assert response.status_code == 200
async def test_stream_iterator(server):
body = b""
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
async with client.stream("GET", server.url) as response:
async for chunk in response.aiter_bytes():
body += chunk
async def test_raw_iterator(server):
body = b""
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
async with client.stream("GET", server.url) as response:
async for chunk in response.aiter_raw():
body += chunk
@pytest.mark.asyncio
async def test_raise_for_status(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
for status_code in (200, 400, 404, 500, 505):
response = await client.request(
"GET", server.url.copy_with(path="/status/{}".format(status_code))
@pytest.mark.asyncio
async def test_options(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.options(server.url)
assert response.status_code == 200
assert response.reason_phrase == "OK"
@pytest.mark.asyncio
async def test_head(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.head(server.url)
assert response.status_code == 200
assert response.reason_phrase == "OK"
@pytest.mark.asyncio
async def test_put(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.put(server.url, data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@pytest.mark.asyncio
async def test_patch(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.patch(server.url, data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@pytest.mark.asyncio
async def test_delete(server):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.delete(server.url)
assert response.status_code == 200
assert response.reason_phrase == "OK"
@pytest.mark.asyncio
async def test_base_url(server):
base_url = server.url
- async with httpx.Client(base_url=base_url) as client:
+ async with httpx.AsyncClient(base_url=base_url) as client:
response = await client.get("/")
assert response.status_code == 200
assert response.url == base_url
url = uds_server.url
uds = uds_server.config.uds
assert uds is not None
- async with httpx.Client(uds=uds) as client:
+ async with httpx.AsyncClient(uds=uds) as client:
response = await client.get(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
def test_merge_url():
- client = httpx.Client(base_url="https://www.paypal.com/")
+ client = httpx.AsyncClient(base_url="https://www.paypal.com/")
url = client.merge_url("http://www.paypal.com")
assert url.scheme == "https"
@pytest.mark.asyncio
async def test_elapsed_delay(server):
url = server.url.copy_with(path="/slow_response/100")
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.get(url)
assert response.elapsed.total_seconds() > 0.0
import pytest
-from httpx import Client, Cookies, Request, Response
+from httpx import AsyncClient, Cookies, Request, Response
from httpx.config import CertTypes, TimeoutTypes, VerifyTypes
from httpx.dispatch.base import Dispatcher
url = "http://example.org/echo_cookies"
cookies = {"example-name": "example-value"}
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url, cookies=cookies)
assert response.status_code == 200
)
cookies.set_cookie(cookie)
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url, cookies=cookies)
assert response.status_code == 200
)
cookies.set_cookie(cookie)
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
client.cookies = cookies # type: ignore
response = await client.get(url)
cookies = Cookies()
cookies["example-name"] = "example-value"
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url, cookies=cookies)
assert response.status_code == 200
async def test_get_cookie() -> None:
url = "http://example.org/set_cookie"
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url)
assert response.status_code == 200
"""
Ensure that Client instances persist cookies between requests.
"""
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get("http://example.org/echo_cookies")
assert response.status_code == 200
import pytest
-from httpx import Client, Headers, Request, Response, __version__
+from httpx import AsyncClient, Headers, Request, Response, __version__
from httpx.config import CertTypes, TimeoutTypes, VerifyTypes
from httpx.dispatch.base import Dispatcher
url = "http://example.org/echo_headers"
headers = {"Example-Header": "example-value"}
- client = Client(dispatch=MockDispatch(), headers=headers)
+ client = AsyncClient(dispatch=MockDispatch(), headers=headers)
response = await client.get(url)
assert response.status_code == 200
url = "http://example.org/echo_headers"
client_headers = {"User-Agent": "python-myclient/0.2.1"}
request_headers = {"X-Auth-Token": "FooBarBazToken"}
- client = Client(dispatch=MockDispatch(), headers=client_headers)
+ client = AsyncClient(dispatch=MockDispatch(), headers=client_headers)
response = await client.get(url, headers=request_headers)
assert response.status_code == 200
url = "http://example.org/echo_headers"
client_headers = {"X-Auth-Token": "FooBar"}
request_headers = {"X-Auth-Token": "BazToken"}
- client = Client(dispatch=MockDispatch(), headers=client_headers)
+ client = AsyncClient(dispatch=MockDispatch(), headers=client_headers)
response = await client.get(url, headers=request_headers)
assert response.status_code == 200
@pytest.mark.asyncio
async def test_header_update():
url = "http://example.org/echo_headers"
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
first_response = await client.get(url)
client.headers.update(
{"User-Agent": "python-myclient/0.2.1", "Another-Header": "AThing"}
"""
url = "http://username:password@example.org:80/echo_headers"
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url)
assert response.status_code == 200
"""
url = "http://username:password@example.org:123/echo_headers"
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get(url)
assert response.status_code == 200
-from httpx import Client, Cookies, Headers
+from httpx import AsyncClient, Cookies, Headers
def test_client_headers():
- client = Client()
+ client = AsyncClient()
client.headers = {"a": "b"}
assert isinstance(client.headers, Headers)
assert client.headers["A"] == "b"
def test_client_cookies():
- client = Client()
+ client = AsyncClient()
client.cookies = {"a": "b"}
assert isinstance(client.cookies, Cookies)
mycookies = list(client.cookies.jar)
],
)
def test_proxies_parameter(proxies, expected_proxies):
- client = httpx.Client(proxies=proxies)
+ client = httpx.AsyncClient(proxies=proxies)
for proxy_key, url in expected_proxies:
assert proxy_key in client.proxies
def test_proxies_has_same_properties_as_dispatch():
- client = httpx.Client(
+ client = httpx.AsyncClient(
proxies="http://127.0.0.1",
verify="/path/to/verify",
cert="/path/to/cert",
],
)
def test_dispatcher_for_request(url, proxies, expected):
- client = httpx.Client(proxies=proxies)
+ client = httpx.AsyncClient(proxies=proxies)
dispatcher = client.dispatcher_for_url(httpx.URL(url))
if expected is None:
def test_unsupported_proxy_scheme():
with pytest.raises(ValueError):
- httpx.Client(proxies="ftp://127.0.0.1")
+ httpx.AsyncClient(proxies="ftp://127.0.0.1")
import pytest
-from httpx import URL, Client, QueryParams, Request, Response
+from httpx import URL, AsyncClient, QueryParams, Request, Response
from httpx.config import CertTypes, TimeoutTypes, VerifyTypes
from httpx.dispatch.base import Dispatcher
def test_client_queryparams():
- client = Client(params={"a": "b"})
+ client = AsyncClient(params={"a": "b"})
assert isinstance(client.params, QueryParams)
assert client.params["a"] == "b"
def test_client_queryparams_string():
- client = Client(params="a=b")
+ client = AsyncClient(params="a=b")
assert isinstance(client.params, QueryParams)
assert client.params["a"] == "b"
- client = Client()
+ client = AsyncClient()
client.params = "a=b"
assert isinstance(client.params, QueryParams)
assert client.params["a"] == "b"
url = "http://example.org/echo_queryparams"
client_queryparams = "first=str"
request_queryparams = {"second": "dict"}
- client = Client(dispatch=MockDispatch(), params=client_queryparams)
+ client = AsyncClient(dispatch=MockDispatch(), params=client_queryparams)
response = await client.get(url, params=request_queryparams)
assert response.status_code == 200
from httpx import (
URL,
- Client,
+ AsyncClient,
NotRedirectResponse,
RedirectBodyUnavailable,
RedirectLoop,
@pytest.mark.usefixtures("async_environment")
async def test_no_redirect():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
url = "https://example.com/no_redirect"
response = await client.get(url)
assert response.status_code == 200
@pytest.mark.usefixtures("async_environment")
async def test_redirect_301():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.post("https://example.org/redirect_301")
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/")
@pytest.mark.usefixtures("async_environment")
async def test_redirect_302():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.post("https://example.org/redirect_302")
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/")
@pytest.mark.usefixtures("async_environment")
async def test_redirect_303():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get("https://example.org/redirect_303")
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/")
@pytest.mark.usefixtures("async_environment")
async def test_disallow_redirects():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.post(
"https://example.org/redirect_303", allow_redirects=False
)
@pytest.mark.usefixtures("async_environment")
async def test_relative_redirect():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get("https://example.org/relative_redirect")
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/")
@pytest.mark.usefixtures("async_environment")
async def test_no_scheme_redirect():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get("https://example.org/no_scheme_redirect")
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/")
@pytest.mark.usefixtures("async_environment")
async def test_fragment_redirect():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get("https://example.org/relative_redirect#fragment")
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/#fragment")
@pytest.mark.usefixtures("async_environment")
async def test_multiple_redirects():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
response = await client.get("https://example.org/multiple_redirects?count=20")
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/multiple_redirects")
@pytest.mark.usefixtures("async_environment")
async def test_too_many_redirects():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
with pytest.raises(TooManyRedirects):
await client.get("https://example.org/multiple_redirects?count=21")
@pytest.mark.usefixtures("async_environment")
async def test_too_many_redirects_calling_next():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
url = "https://example.org/multiple_redirects?count=21"
response = await client.get(url, allow_redirects=False)
with pytest.raises(TooManyRedirects):
@pytest.mark.usefixtures("async_environment")
async def test_redirect_loop():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
with pytest.raises(RedirectLoop):
await client.get("https://example.org/redirect_loop")
@pytest.mark.usefixtures("async_environment")
async def test_redirect_loop_calling_next():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
url = "https://example.org/redirect_loop"
response = await client.get(url, allow_redirects=False)
with pytest.raises(RedirectLoop):
@pytest.mark.usefixtures("async_environment")
async def test_cross_domain_redirect():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
url = "https://example.com/cross_domain"
headers = {"Authorization": "abc"}
response = await client.get(url, headers=headers)
@pytest.mark.usefixtures("async_environment")
async def test_same_domain_redirect():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
url = "https://example.org/cross_domain"
headers = {"Authorization": "abc"}
response = await client.get(url, headers=headers)
"""
A 308 redirect should preserve the request body.
"""
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
url = "https://example.org/redirect_body"
data = b"Example request body"
response = await client.post(url, data=data)
"""
A 303 redirect should remove the request body.
"""
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
url = "https://example.org/redirect_no_body"
data = b"Example request body"
response = await client.post(url, data=data)
@pytest.mark.usefixtures("async_environment")
async def test_cannot_redirect_streaming_body():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
url = "https://example.org/redirect_body"
async def streaming_body():
@pytest.mark.usefixtures("async_environment")
async def test_cross_subdomain_redirect():
- client = Client(dispatch=MockDispatch())
+ client = AsyncClient(dispatch=MockDispatch())
url = "https://example.com/cross_subdomain"
response = await client.get(url)
assert response.url == URL("https://www.example.org/cross_subdomain")
@pytest.mark.usefixtures("async_environment")
async def test_redirect_cookie_behavior():
- client = Client(dispatch=MockCookieDispatch())
+ client = AsyncClient(dispatch=MockCookieDispatch())
# The client is not logged in.
response = await client.get("https://example.com/")
import pytest
from h2.settings import SettingCodes
-from httpx import Client, Response, TimeoutException
+from httpx import AsyncClient, Response, TimeoutException
from .utils import MockHTTP2Backend
async def test_http2_get_request():
backend = MockHTTP2Backend(app=app)
- async with Client(backend=backend, http2=True) as client:
+ async with AsyncClient(backend=backend, http2=True) as client:
response = await client.get("http://example.org")
assert response.status_code == 200
async def test_http2_post_request():
backend = MockHTTP2Backend(app=app)
- async with Client(backend=backend, http2=True) as client:
+ async with AsyncClient(backend=backend, http2=True) as client:
response = await client.post("http://example.org", data=b"<data>")
assert response.status_code == 200
backend = MockHTTP2Backend(app=app)
data = b"a" * 100000
- async with Client(backend=backend, http2=True) as client:
+ async with AsyncClient(backend=backend, http2=True) as client:
response = await client.post("http://example.org", data=data)
assert response.status_code == 200
assert json.loads(response.content) == {
async def test_http2_multiple_requests():
backend = MockHTTP2Backend(app=app)
- async with Client(backend=backend, http2=True) as client:
+ async with AsyncClient(backend=backend, http2=True) as client:
response_1 = await client.get("http://example.org/1")
response_2 = await client.get("http://example.org/2")
response_3 = await client.get("http://example.org/3")
"""
backend = MockHTTP2Backend(app=app)
- async with Client(backend=backend, http2=True) as client:
+ async with AsyncClient(backend=backend, http2=True) as client:
response_1 = await client.get("http://example.org/1")
backend.server.close_connection = True
response_2 = await client.get("http://example.org/2")
async def test_http2_settings_in_handshake():
backend = MockHTTP2Backend(app=app)
- async with Client(backend=backend, http2=True) as client:
+ async with AsyncClient(backend=backend, http2=True) as client:
await client.get("http://example.org")
h2_conn = backend.server.conn
@pytest.mark.usefixtures("async_environment")
async def test_http2_live_request():
- async with Client(http2=True) as client:
+ async with AsyncClient(http2=True) as client:
try:
resp = await client.get("https://nghttp2.org/httpbin/anything")
except TimeoutException:
@pytest.mark.asyncio
async def test_asgi():
- client = httpx.Client(app=hello_world)
+ client = httpx.AsyncClient(app=hello_world)
response = await client.get("http://www.example.org/")
assert response.status_code == 200
assert response.text == "Hello, World!"
@pytest.mark.asyncio
async def test_asgi_upload():
- client = httpx.Client(app=echo_body)
+ client = httpx.AsyncClient(app=echo_body)
response = await client.post("http://www.example.org/", data=b"example")
assert response.status_code == 200
assert response.text == "example"
@pytest.mark.asyncio
async def test_asgi_exc():
- client = httpx.Client(app=raise_exc)
+ client = httpx.AsyncClient(app=raise_exc)
with pytest.raises(ValueError):
await client.get("http://www.example.org/")
@pytest.mark.asyncio
async def test_asgi_exc_after_response():
- client = httpx.Client(app=raise_exc_after_response)
+ client = httpx.AsyncClient(app=raise_exc_after_response)
with pytest.raises(ValueError):
await client.get("http://www.example.org/")
@pytest.mark.parametrize(("value,output"), (("abc", b"abc"), (b"abc", b"abc")))
@pytest.mark.asyncio
async def test_multipart(value, output):
- client = httpx.Client(dispatch=MockDispatch())
+ client = httpx.AsyncClient(dispatch=MockDispatch())
# Test with a single-value 'data' argument, and a plain file 'files' argument.
data = {"text": value}
@pytest.mark.parametrize(("key"), (b"abc", 1, 2.3, None))
@pytest.mark.asyncio
async def test_multipart_invalid_key(key):
- client = httpx.Client(dispatch=MockDispatch())
+ client = httpx.AsyncClient(dispatch=MockDispatch())
data = {key: "abc"}
files = {"file": io.BytesIO(b"<file content>")}
with pytest.raises(TypeError) as e:
@pytest.mark.parametrize(("value"), (1, 2.3, None, [None, "abc"], {None: "abc"}))
@pytest.mark.asyncio
async def test_multipart_invalid_value(value):
- client = httpx.Client(dispatch=MockDispatch())
+ client = httpx.AsyncClient(dispatch=MockDispatch())
data = {"text": value}
files = {"file": io.BytesIO(b"<file content>")}
with pytest.raises(TypeError) as e:
@pytest.mark.asyncio
async def test_multipart_file_tuple():
- client = httpx.Client(dispatch=MockDispatch())
+ client = httpx.AsyncClient(dispatch=MockDispatch())
# Test with a list of values 'data' argument, and a tuple style 'files' argument.
data = {"text": ["abc"]}
async def test_read_timeout(server):
timeout = httpx.Timeout(read_timeout=1e-6)
- async with httpx.Client(timeout=timeout) as client:
+ async with httpx.AsyncClient(timeout=timeout) as client:
with pytest.raises(httpx.ReadTimeout):
await client.get(server.url.copy_with(path="/slow_response"))
async def test_write_timeout(server):
timeout = httpx.Timeout(write_timeout=1e-6)
- async with httpx.Client(timeout=timeout) as client:
+ async with httpx.AsyncClient(timeout=timeout) as client:
with pytest.raises(httpx.WriteTimeout):
data = b"*" * 1024 * 1024 * 100
await client.put(server.url.copy_with(path="/slow_response"), data=data)
async def test_connect_timeout(server):
timeout = httpx.Timeout(connect_timeout=1e-6)
- async with httpx.Client(timeout=timeout) as client:
+ async with httpx.AsyncClient(timeout=timeout) as client:
with pytest.raises(httpx.ConnectTimeout):
# See https://stackoverflow.com/questions/100841/
await client.get("http://10.255.255.1/")
pool_limits = httpx.PoolLimits(hard_limit=1)
timeout = httpx.Timeout(pool_timeout=1e-4)
- async with httpx.Client(pool_limits=pool_limits, timeout=timeout) as client:
+ async with httpx.AsyncClient(pool_limits=pool_limits, timeout=timeout) as client:
async with client.stream("GET", server.url):
with pytest.raises(httpx.PoolTimeout):
await client.get("http://localhost:8000/")
@pytest.mark.asyncio
async def test_logs_debug(server, capsys):
with override_log_level("debug"):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.get(server.url)
assert response.status_code == 200
stderr = capsys.readouterr().err
@pytest.mark.asyncio
async def test_logs_trace(server, capsys):
with override_log_level("trace"):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.get(server.url)
assert response.status_code == 200
stderr = capsys.readouterr().err
@pytest.mark.asyncio
async def test_logs_redirect_chain(server, capsys):
with override_log_level("debug"):
- async with httpx.Client() as client:
+ async with httpx.AsyncClient() as client:
response = await client.get(server.url.copy_with(path="/redirect_301"))
assert response.status_code == 200
projects / thirdparty / httpx.git / commitdiff
