-# HTTP3
+# HTTPX
-<a href="https://travis-ci.org/encode/http3">
- <img src="https://travis-ci.org/encode/http3.svg?branch=master" alt="Build Status">
+<a href="https://travis-ci.org/encode/httpx">
+ <img src="https://travis-ci.org/encode/httpx.svg?branch=master" alt="Build Status">
</a>
-<a href="https://codecov.io/gh/encode/http3">
- <img src="https://codecov.io/gh/encode/http3/branch/master/graph/badge.svg" alt="Coverage">
+<a href="https://codecov.io/gh/encode/httpx">
+ <img src="https://codecov.io/gh/encode/httpx/branch/master/graph/badge.svg" alt="Coverage">
</a>
-<a href="https://pypi.org/project/http3/">
- <img src="https://badge.fury.io/py/http3.svg" alt="Package version">
+<a href="https://pypi.org/project/httpx/">
+ <img src="https://badge.fury.io/py/httpx.svg" alt="Package version">
</a>
-HTTP3 is a next-generation HTTP client for Python 3.
+HTTPX is a next-generation HTTP client for Python 3.
**Note**: *This project should be considered as an "alpha" release. It is substantially API complete, but there are still some areas that need more work.*
Let's get started...
```python
->>> import http3
->>> r = http3.get('https://www.example.org/')
+>>> import httpx
+>>> r = httpx.get('https://www.example.org/')
>>> r
<Response [200 OK]>
>>> r.status_code
## Features
-HTTP3 builds on the well-established usability of `requests`, and gives you:
+HTTPX builds on the well-established usability of `requests`, and gives you:
* A requests-compatible API.
* HTTP/2 and HTTP/1.1 support.
-* Support for [issuing HTTP requests in parallel](https://www.encode.io/http3/parallel/). *(Coming soon)*
-* Standard synchronous interface, but [with `async`/`await` support if you need it](https://www.encode.io/http3/async/).
-* Ability to [make requests directly to WSGI or ASGI applications](https://www.encode.io/http3/advanced/#calling-into-python-web-apps).
+* Support for [issuing HTTP requests in parallel](https://www.encode.io/httpx/parallel/). *(Coming soon)*
+* Standard synchronous interface, but [with `async`/`await` support if you need it](https://www.encode.io/httpx/async/).
+* Ability to [make requests directly to WSGI or ASGI applications](https://www.encode.io/httpx/advanced/#calling-into-python-web-apps).
* Strict timeouts everywhere.
* Fully type annotated.
* 100% test coverage.
Install with pip:
```shell
-$ pip install http3
+$ pip install httpx
```
-HTTP3 requires Python 3.6+
+httpx requires Python 3.6+
## Documentation
-Project documentation is available at [www.encode.io/http3/](https://www.encode.io/http3/).
+Project documentation is available at [www.encode.io/httpx/](https://www.encode.io/httpx/).
-For a run-through of all the basics, head over to the [QuickStart](https://www.encode.io/http3/quickstart/).
+For a run-through of all the basics, head over to the [QuickStart](https://www.encode.io/httpx/quickstart/).
-For more advanced topics, see the [Advanced Usage](https://www.encode.io/http3/advanced/) section, or
-the specific topics on making [Parallel Requests](https://www.encode.io/http3/parallel/) or using the
-[Async Client](https://www.encode.io/http3/async/).
+For more advanced topics, see the [Advanced Usage](https://www.encode.io/httpx/advanced/) section, or
+the specific topics on making [Parallel Requests](https://www.encode.io/httpx/parallel/) or using the
+[Async Client](https://www.encode.io/httpx/async/).
-The [Developer Interface](https://www.encode.io/http3/api/) provides a comprehensive API reference.
+The [Developer Interface](https://www.encode.io/httpx/api/) provides a comprehensive API reference.
## Dependencies
-The HTTP3 project relies on these excellent libraries:
+The httpx project relies on these excellent libraries:
* `h2` - HTTP/2 support.
* `h11` - HTTP/1.1 support.
inspiration around the lower level networking details.
<p align="center">— ⭐️ —</p>
-<p align="center"><i>HTTP3 is <a href="https://github.com/encode/http3/blob/master/LICENSE.md">BSD licensed</a> code. Designed & built in Brighton, England.</i></p>
+<p align="center"><i>httpx is <a href="https://github.com/encode/httpx/blob/master/LICENSE.md">BSD licensed</a> code. Designed & built in Brighton, England.</i></p>
all outgoing requests.
```python
->>> client = http3.Client()
+>>> client = httpx.Client()
>>> r = client.get('https://example.org/')
>>> r
<Response [200 OK]>
## Calling into Python Web Apps
-You can configure an `http3` client to call directly into a Python web
+You can configure an `httpx` client to call directly into a Python web
application, using either the WSGI or ASGI protocol.
This is particularly useful for two main use-cases:
-* Using `http3` as a client, inside test cases.
+* Using `httpx` as a client, inside test cases.
* Mocking out external services, during tests or in dev/staging environments.
Here's an example of integrating against a Flask application:
```python
from flask import Flask
-import http3
+import httpx
app = Flask(__name__)
def hello():
return "Hello World!"
-client = http3.Client(app=app)
+client = httpx.Client(app=app)
r = client.get('http://example/')
assert r.status_code == 200
assert r.text == "Hello World!"
*An HTTP client, with connection pooling, redirects, cookie persistence, etc.*
```python
->>> client = http3.Client()
+>>> client = httpx.Client()
>>> response = client.get('https://example.org')
```
what gets sent over the wire.*
```python
->>> request = http3.Request("GET", "https://example.org", headers={'host': 'example.org'})
+>>> request = httpx.Request("GET", "https://example.org", headers={'host': 'example.org'})
>>> response = client.send(request)
```
# Async Client
-HTTP3 offers a standard synchronous API by default, but also gives you
+HTTPX offers a standard synchronous API by default, but also gives you
the option of an async client if you need it.
Async is a concurrency model that is far more efficient than multi-threading,
To make asynchronous requests, you'll need an `AsyncClient`.
```python
->>> client = http3.AsyncClient()
+>>> client = httpx.AsyncClient()
>>> r = await client.get('https://www.example.com/')
```
use async methods:
```python
->>> client = http3.AsyncClient()
+>>> client = httpx.AsyncClient()
>>> r = await client.get('https://www.example.com/', stream=True)
>>> try:
>>> async for chunk in r.stream():
If you're making parallel requests, then you'll also need to use an async API:
```python
->>> client = http3.AsyncClient()
+>>> client = httpx.AsyncClient()
>>> async with client.parallel() as parallel:
>>> pending_one = parallel.get('https://example.com/1')
>>> pending_two = parallel.get('https://example.com/2')
# Requests Compatibility Guide
-HTTP3 aims to be compatible with the `requests` API wherever possible.
+HTTPX aims to be compatible with the `requests` API wherever possible.
This documentation outlines places where the API differs...
to the API in our own documentation. The following exceptions apply:
* `Response.url` - Returns a `URL` instance, rather than a string. Use `str(response.url)` if you need a string instance.
-* `http3.codes` - In our documentation we prefer the uppercased versions, such as `codes.NOT_FOUND`,
+* `httpx.codes` - In our documentation we prefer the uppercased versions, such as `codes.NOT_FOUND`,
but also provide lower-cased versions for API compatibility with `requests`.
* `stream=True`. - Streaming responses provide the `.stream()` and `.raw()` byte iterator interfaces, rather than the `.iter_content()` method and the `.raw` socket interface.
-<h1 style="margin: 0">HTTP3</h1>
+<p align="center" style="margin: 0 0 10px">
+ <img width="350" height="208" src="http://127.0.0.1:8000/img/logo.jpg" alt='HTTPX'>
+</p>
+<h1 align="center" style="font-size: 3rem; margin: -15px 0">
+HTTPX
+</h1>
+
+---
+
+<div align="center">
+<p>
<a href="https://travis-ci.org/encode/http3">
<img src="https://travis-ci.org/encode/http3.svg?branch=master" alt="Build Status">
</a>
<a href="https://pypi.org/project/http3/">
<img src="https://badge.fury.io/py/http3.svg" alt="Package version">
</a>
+</p>
-HTTP3 is a next-generation HTTP client for Python 3.
+<em>A next-generation HTTP client for Python.</em>
+</div>
!!! warning
This project should be considered as an "alpha" release. It is substantially
Let's get started...
```python
->>> import http3
->>> r = http3.get('https://www.example.org/')
+>>> import httpx
+>>> r = httpx.get('https://www.example.org/')
>>> r
<Response [200 OK]>
>>> r.status_code
## Features
-HTTP3 builds on the well-established usability of `requests`, and gives you:
+HTTPX builds on the well-established usability of `requests`, and gives you:
* A requests-compatible API.
* HTTP/2 and HTTP/1.1 support.
## Dependencies
-The HTTP3 project relies on these excellent libraries:
+The HTTPX project relies on these excellent libraries:
* `h2` - HTTP/2 support.
* `h11` - HTTP/1.1 support.
Install with pip:
```shell
-$ pip install http3
+$ pip install httpx
```
-HTTP3 requires Python 3.6+
+HTTPX requires Python 3.6+
!!! warning
This page documents some proposed functionality that is not yet released.
- See [pull request #52](https://github.com/encode/http3/pull/52) for the
+ See [pull request #52](https://github.com/encode/httpx/pull/52) for the
first-pass of an implementation.
-HTTP3 allows you to make HTTP requests in parallel in a highly efficient way,
+HTTPX allows you to make HTTP requests in parallel in a highly efficient way,
using async under the hood, while still presenting a standard threaded interface.
This has the huge benefit of allowing you to efficiently make parallel HTTP
Let's make two outgoing HTTP requests in parallel:
```python
->>> with http3.parallel() as parallel:
+>>> with httpx.parallel() as parallel:
>>> pending_one = parallel.get('https://example.com/1')
>>> pending_two = parallel.get('https://example.com/2')
>>> response_one = pending_one.get_response()
as soon as it's available:
```python
->>> with http3.parallel() as parallel:
+>>> with httpx.parallel() as parallel:
>>> for counter in range(1, 10):
>>> parallel.get(f'https://example.com/{counter}')
>>> while parallel.has_pending_responses:
block.
```python
->>> client = http3.Client()
+>>> client = httpx.Client()
>>> with client.parallel() as parallel:
>>> ...
```
async API for making requests.
```python
->>> client = http3.AsyncClient()
+>>> client = httpx.AsyncClient()
>>> async with client.parallel() as parallel:
>>> pending_one = await parallel.get('https://example.com/1')
>>> pending_two = await parallel.get('https://example.com/2')
!!! note
This page closely follows the layout of the `requests` QuickStart documentation.
- The `http3` library is designed to be API compatible with `requests` wherever
+ The `httpx` library is designed to be API compatible with `requests` wherever
possible.
-First start by importing HTTP3:
+First start by importing HTTPX:
```
->>> import http3
+>>> import httpx
```
Now, let’s try to get a webpage.
```python
->>> r = http3.get('https://httpbin.org/get')
+>>> r = httpx.get('https://httpbin.org/get')
>>> r
<Response [200 OK]>
```
Similarly, to make an HTTP POST request:
```python
->>> r = http3.post('https://httpbin.org/post', data={'key': 'value'})
+>>> r = httpx.post('https://httpbin.org/post', data={'key': 'value'})
```
The PUT, DELETE, HEAD, and OPTIONS requests all follow the same style:
```python
->>> r = http3.put('https://httpbin.org/put', data={'key': 'value'})
->>> r = http3.delete('https://httpbin.org/delete')
->>> r = http3.head('https://httpbin.org/get')
->>> r = http3.options('https://httpbin.org/get')
+>>> r = httpx.put('https://httpbin.org/put', data={'key': 'value'})
+>>> r = httpx.delete('https://httpbin.org/delete')
+>>> r = httpx.head('https://httpbin.org/get')
+>>> r = httpx.options('https://httpbin.org/get')
```
## Passing Parameters in URLs
```python
>>> params = {'key1': 'value1', 'key2': 'value2'}
->>> r = http3.get('https://httpbin.org/get', params=params)
+>>> r = httpx.get('https://httpbin.org/get', params=params)
```
To see how the values get encoding into the URL string, we can inspect the
```python
>>> params = {'key1': 'value1', 'key2': ['value2', 'value3']}
->>> r = http3.get('https://httpbin.org/get', params=params)
+>>> r = httpx.get('https://httpbin.org/get', params=params)
>>> r.url
URL('https://httpbin.org/get?key1=value1&key2=value2&key2=value3')
```
## Response Content
-HTTP3 will automatically handle decoding the response content into unicode text.
+HTTPX will automatically handle decoding the response content into unicode text.
```python
->>> r = http3.get('https://www.example.org/')
+>>> r = httpx.get('https://www.example.org/')
>>> r.text
'<!doctype html>\n<html>\n<head>\n<title>Example Domain</title>...'
```
Often Web API responses will be encoded as JSON.
```python
->>> r = http3.get('https://api.github.com/events')
+>>> r = httpx.get('https://api.github.com/events')
>>> r.json()
[{u'repository': {u'open_issues': 0, u'url': 'https://github.com/...' ... }}]
```
```python
>>> url = 'http://httpbin.org/headers'
>>> headers = {'user-agent': 'my-app/0.0.1'}
->>> r = http3.get(url, headers=headers)
+>>> r = httpx.get(url, headers=headers)
```
## Sending Form Encoded Data
```python
>>> data = {'key1': 'value1', 'key2': 'value2'}
->>> r = http3.post("https://httpbin.org/post", data=data)
+>>> r = httpx.post("https://httpbin.org/post", data=data)
>>> print(r.text)
{
...
```python
>>> data = {'key1': ['value1', 'value2']}
->>> r = http3.post("https://httpbin.org/post", data=data)
+>>> r = httpx.post("https://httpbin.org/post", data=data)
>>> print(r.text)
{
...
```python
>>> files = {'upload-file': open('report.xls', 'rb')}
->>> r = http3.post("https://httpbin.org/post", files=files)
+>>> r = httpx.post("https://httpbin.org/post", files=files)
>>> print(r.text)
{
...
```python
>>> files = {'upload-file': ('report.xls', open('report.xls', 'rb'), 'application/vnd.ms-excel')}
->>> r = http3.post("https://httpbin.org/post", files=files)
+>>> r = httpx.post("https://httpbin.org/post", files=files)
>>> print(r.text)
{
...
```python
>>> data = {'integer': 123, 'boolean': True, 'list': ['a', 'b', 'c']}
->>> r = http3.post("https://httpbin.org/post", json=data)
+>>> r = httpx.post("https://httpbin.org/post", json=data)
>>> print(r.text)
{
...
We can inspect the HTTP status code of the response:
```python
->>> r = http3.get('https://httpbin.org/get')
+>>> r = httpx.get('https://httpbin.org/get')
>>> r.status_code
200
```
-HTTP3 also includes an easy shortcut for accessing status codes by their text phrase.
+HTTPX also includes an easy shortcut for accessing status codes by their text phrase.
```python
->>> r.status_code == http3.codes.OK
+>>> r.status_code == httpx.codes.OK
True
```
We can raise an exception for any Client or Server error responses (4xx or 5xx status codes):
```python
->>> not_found = http3.get('https://httpbin.org/status/404')
+>>> not_found = httpx.get('https://httpbin.org/status/404')
>>> not_found.status_code
404
>>> not_found.raise_for_status()
Traceback (most recent call last):
- File "/Users/tomchristie/GitHub/encode/httpcore/http3/models.py", line 776, in raise_for_status
+ File "/Users/tomchristie/GitHub/encode/httpcore/httpx/models.py", line 776, in raise_for_status
raise HttpError(message)
-http3.exceptions.HttpError: 404 Not Found
+httpx.exceptions.HttpError: 404 Not Found
```
Any successful response codes will simply return `None` rather than raising an exception.
Any cookies that are set on the response can be easily accessed:
```python
->>> r = http3.get('http://httpbin.org/cookies/set?chocolate=chip', allow_redirects=False)
+>>> r = httpx.get('http://httpbin.org/cookies/set?chocolate=chip', allow_redirects=False)
>>> r.cookies['chocolate']
'chip'
```
```python
>>> cookies = {"peanut": "butter"}
->>> r = http3.get('http://httpbin.org/cookies', cookies=cookies)
+>>> r = httpx.get('http://httpbin.org/cookies', cookies=cookies)
>>> r.json()
{'cookies': {'peanut': 'butter'}}
```
with additional API for accessing cookies by their domain or path.
```python
->>> cookies = http3.Cookies()
+>>> cookies = httpx.Cookies()
>>> cookies.set('cookie_on_domain', 'hello, there!', domain='httpbin.org')
>>> cookies.set('cookie_off_domain', 'nope.', domain='example.org')
->>> r = http3.get('http://httpbin.org/cookies', cookies=cookies)
+>>> r = httpx.get('http://httpbin.org/cookies', cookies=cookies)
>>> r.json()
{'cookies': {'cookie_on_domain': 'hello, there!'}}
```
## Redirection and History
-By default HTTP3 will follow redirects for anything except `HEAD` requests.
+By default HTTPX will follow redirects for anything except `HEAD` requests.
The `history` property of the response can be used to inspect any followed redirects.
It contains a list of all any redirect responses that were followed, in the order
For example, GitHub redirects all HTTP requests to HTTPS.
```python
->>> r = http3.get('http://github.com/')
+>>> r = httpx.get('http://github.com/')
>>> r.url
URL('https://github.com/')
>>> r.status_code
You can modify the default redirection handling with the allow_redirects parameter:
```python
->>> r = http3.get('http://github.com/', allow_redirects=False)
+>>> r = httpx.get('http://github.com/', allow_redirects=False)
>>> r.status_code
301
>>> r.history
If you’re making a `HEAD` request, you can use this to enable redirection:
```python
->>> r = http3.head('http://github.com/', allow_redirects=True)
+>>> r = httpx.head('http://github.com/', allow_redirects=True)
>>> r.url
'https://github.com/'
>>> r.history
## Timeouts
-HTTP3 defaults to including reasonable timeouts for all network operations,
+HTTPX defaults to including reasonable timeouts for all network operations,
meaning that if a connection is not properly established then it should always
raise an error rather than hanging indefinitely.
value to be more or less strict:
```python
->>> http3.get('https://github.com/', timeout=0.001)
+>>> httpx.get('https://github.com/', timeout=0.001)
```
-__title__ = "http3"
+__title__ = "httpx"
__description__ = "A next generation HTTP client, for Python 3."
__version__ = "0.6.7"
TimeoutTypes = typing.Union[float, typing.Tuple[float, float, float], "TimeoutConfig"]
-USER_AGENT = f"python-http3/{__version__}"
+USER_AGENT = f"python-httpx/{__version__}"
DEFAULT_CIPHERS = ":".join(
[
and will setup an appropriate dispatch class:
```
- client = http3.Client(app=app)
+ client = httpx.Client(app=app)
```
Alternatively, you can setup the dispatch instance explicitly.
to the ASGIDispatch class:
```
- dispatch = http3.ASGIDispatch(
+ dispatch = httpx.ASGIDispatch(
app=app,
root_path="/submount",
client=("1.2.3.4", 123)
)
- client = http3.Client(dispatch=dispatch)
+ client = httpx.Client(dispatch=dispatch)
```
"""
and will setup an appropriate dispatch class:
```
- client = http3.Client(app=app)
+ client = httpx.Client(app=app)
```
Alternatively, you can setup the dispatch instance explicitly.
to the WSGIDispatch class:
```
- dispatch = http3.WSGIDispatch(
+ dispatch = httpx.WSGIDispatch(
app=app,
script_name="/submount",
remote_addr="1.2.3.4"
)
- client = http3.Client(dispatch=dispatch)
+ client = httpx.Client(dispatch=dispatch)
```
"""
-site_name: HTTP3
-site_description: The next generation HTTP client.
+site_name: HTTPX
+site_description: A next-generation HTTP client for Python.
theme:
name: 'material'
-repo_name: encode/http3
-repo_url: https://github.com/encode/http3
+repo_name: encode/httpx
+repo_url: https://github.com/encode/httpx
edit_uri: ""
nav:
if [ -d 'htmlcov' ] ; then
rm -r htmlcov
fi
-if [ -d 'http3.egg-info' ] ; then
- rm -r http3.egg-info
+if [ -d 'httpx.egg-info' ] ; then
+ rm -r httpx.egg-info
fi
set -x
-${PREFIX}autoflake --in-place --recursive http3 tests setup.py
-${PREFIX}isort --multi-line=3 --trailing-comma --force-grid-wrap=0 --combine-as --line-width 88 --recursive --apply http3 tests setup.py
-${PREFIX}black http3 tests setup.py
-${PREFIX}flake8 --max-line-length=88 --ignore=W503,E203 http3 tests setup.py
-${PREFIX}mypy http3 --ignore-missing-imports --disallow-untyped-defs
+${PREFIX}autoflake --in-place --recursive httpx tests setup.py
+${PREFIX}isort --multi-line=3 --trailing-comma --force-grid-wrap=0 --combine-as --line-width 88 --recursive --apply httpx tests setup.py
+${PREFIX}black httpx tests setup.py
+${PREFIX}flake8 --max-line-length=88 --ignore=W503,E203 httpx tests setup.py
+${PREFIX}mypy httpx --ignore-missing-imports --disallow-untyped-defs
scripts/clean
#!/bin/sh -e
-export PACKAGE="http3"
+export PACKAGE="httpx"
export VERSION=`cat ${PACKAGE}/__version__.py | grep __version__ | sed "s/__version__ = //" | sed "s/'//g"`
export PREFIX=""
if [ -d 'venv' ] ; then
#!/bin/sh -e
-export PACKAGE="http3"
+export PACKAGE="httpx"
export PREFIX=""
if [ -d 'venv' ] ; then
export PREFIX="venv/bin/"
import pytest
-import http3
+import httpx
@pytest.mark.asyncio
async def test_get(server):
url = "http://127.0.0.1:8000/"
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.get(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
@pytest.mark.asyncio
async def test_post(server):
url = "http://127.0.0.1:8000/"
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.post(url, data=b"Hello, world!")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_post_json(server):
url = "http://127.0.0.1:8000/"
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.post(url, json={"text": "Hello, world!"})
assert response.status_code == 200
@pytest.mark.asyncio
async def test_stream_response(server):
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.request("GET", "http://127.0.0.1:8000/", stream=True)
assert response.status_code == 200
body = await response.read()
@pytest.mark.asyncio
async def test_access_content_stream_response(server):
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.request("GET", "http://127.0.0.1:8000/", stream=True)
assert response.status_code == 200
- with pytest.raises(http3.ResponseNotRead):
+ with pytest.raises(httpx.ResponseNotRead):
response.content
yield b"Hello, "
yield b"world!"
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.request(
"POST", "http://127.0.0.1:8000/", data=hello_world()
)
@pytest.mark.asyncio
async def test_raise_for_status(server):
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
for status_code in (200, 400, 404, 500, 505):
response = await client.request(
"GET", f"http://127.0.0.1:8000/status/{status_code}"
)
if 400 <= status_code < 600:
- with pytest.raises(http3.exceptions.HttpError):
+ with pytest.raises(httpx.exceptions.HttpError):
response.raise_for_status()
else:
assert response.raise_for_status() is None
@pytest.mark.asyncio
async def test_options(server):
url = "http://127.0.0.1:8000/"
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.options(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
@pytest.mark.asyncio
async def test_head(server):
url = "http://127.0.0.1:8000/"
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.head(url)
assert response.status_code == 200
assert response.text == ""
@pytest.mark.asyncio
async def test_put(server):
url = "http://127.0.0.1:8000/"
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.put(url, data=b"Hello, world!")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_patch(server):
url = "http://127.0.0.1:8000/"
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.patch(url, data=b"Hello, world!")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_delete(server):
url = "http://127.0.0.1:8000/"
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.delete(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
headers = {"Expect": "100-continue"}
data = b"Echo request body"
- async with http3.AsyncClient() as client:
+ async with httpx.AsyncClient() as client:
response = await client.post(url, headers=headers, data=data)
assert response.status_code == 200
import json
-from http3 import (
+from httpx import (
AsyncDispatcher,
AsyncRequest,
AsyncResponse,
import pytest
-import http3
+import httpx
def threadpool(func):
@threadpool
def test_get(server):
url = "http://127.0.0.1:8000/"
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.get(url)
assert response.status_code == 200
- assert response.url == http3.URL(url)
+ assert response.url == httpx.URL(url)
assert response.content == b"Hello, world!"
assert response.text == "Hello, world!"
assert response.protocol == "HTTP/1.1"
assert response.encoding == "iso-8859-1"
- assert response.request.url == http3.URL(url)
+ assert response.request.url == httpx.URL(url)
assert response.headers
assert response.is_redirect is False
assert repr(response) == "<Response [200 OK]>"
@threadpool
def test_post(server):
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.post("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_post_json(server):
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.post("http://127.0.0.1:8000/", json={"text": "Hello, world!"})
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_stream_response(server):
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.get("http://127.0.0.1:8000/", stream=True)
assert response.status_code == 200
content = response.read()
@threadpool
def test_stream_iterator(server):
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.get("http://127.0.0.1:8000/", stream=True)
assert response.status_code == 200
body = b""
@threadpool
def test_raw_iterator(server):
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.get("http://127.0.0.1:8000/", stream=True)
assert response.status_code == 200
body = b""
@threadpool
def test_raise_for_status(server):
- with http3.Client() as client:
+ with httpx.Client() as client:
for status_code in (200, 400, 404, 500, 505):
response = client.request(
"GET", "http://127.0.0.1:8000/status/{}".format(status_code)
)
if 400 <= status_code < 600:
- with pytest.raises(http3.exceptions.HttpError):
+ with pytest.raises(httpx.exceptions.HttpError):
response.raise_for_status()
else:
assert response.raise_for_status() is None
@threadpool
def test_options(server):
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.options("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_head(server):
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.head("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_put(server):
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.put("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_patch(server):
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.patch("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_delete(server):
- with http3.Client() as http:
+ with httpx.Client() as http:
response = http.delete("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_base_url(server):
base_url = "http://127.0.0.1:8000/"
- with http3.Client(base_url=base_url) as http:
+ with httpx.Client(base_url=base_url) as http:
response = http.get("/")
assert response.status_code == 200
assert str(response.url) == base_url
import json
from http.cookiejar import Cookie, CookieJar
-from http3 import (
+from httpx import (
AsyncDispatcher,
AsyncRequest,
AsyncResponse,
import pytest
-from http3 import (
+from httpx import (
URL,
AsyncClient,
AsyncDispatcher,
import pytest
-import http3
+import httpx
@pytest.mark.asyncio
"""
Connections should default to staying in a keep-alive state.
"""
- async with http3.ConnectionPool() as http:
+ async with httpx.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
await response.read()
assert len(http.active_connections) == 0
"""
Connnections to differing connection keys should result in multiple connections.
"""
- async with http3.ConnectionPool() as http:
+ async with httpx.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
await response.read()
assert len(http.active_connections) == 0
"""
The soft_limit config should limit the maximum number of keep-alive connections.
"""
- pool_limits = http3.PoolLimits(soft_limit=1)
+ pool_limits = httpx.PoolLimits(soft_limit=1)
- async with http3.ConnectionPool(pool_limits=pool_limits) as http:
+ async with httpx.ConnectionPool(pool_limits=pool_limits) as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
await response.read()
assert len(http.active_connections) == 0
"""
A streaming request should hold the connection open until the response is read.
"""
- async with http3.ConnectionPool() as http:
+ async with httpx.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
assert len(http.active_connections) == 1
assert len(http.keepalive_connections) == 0
"""
Multiple conncurrent requests should open multiple conncurrent connections.
"""
- async with http3.ConnectionPool() as http:
+ async with httpx.ConnectionPool() as http:
response_a = await http.request("GET", "http://127.0.0.1:8000/")
assert len(http.active_connections) == 1
assert len(http.keepalive_connections) == 0
Using a `Connection: close` header should close the connection.
"""
headers = [(b"connection", b"close")]
- async with http3.ConnectionPool() as http:
+ async with httpx.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/", headers=headers)
await response.read()
assert len(http.active_connections) == 0
"""
A standard close should keep the connection open.
"""
- async with http3.ConnectionPool() as http:
+ async with httpx.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
await response.read()
await response.close()
"""
A premature close should close the connection.
"""
- async with http3.ConnectionPool() as http:
+ async with httpx.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
await response.close()
assert len(http.active_connections) == 0
import pytest
-from http3 import HTTPConnection
+from httpx import HTTPConnection
@pytest.mark.asyncio
import json
-from http3 import Client, Response
+from httpx import Client, Response
from .utils import MockHTTP2Backend
import json
-from http3 import (
+from httpx import (
CertTypes,
Client,
Dispatcher,
import h2.connection
import h2.events
-from http3 import (
+from httpx import (
AsyncioBackend,
BaseReader,
BaseWriter,
import pytest
-from http3 import CookieConflict, Cookies
+from httpx import CookieConflict, Cookies
def test_cookies():
-import http3
+import httpx
def test_headers():
- h = http3.Headers([("a", "123"), ("a", "456"), ("b", "789")])
+ h = httpx.Headers([("a", "123"), ("a", "456"), ("b", "789")])
assert "a" in h
assert "A" in h
assert "b" in h
assert list(h) == ["a", "a", "b"]
assert dict(h) == {"a": "123, 456", "b": "789"}
assert repr(h) == "Headers([('a', '123'), ('a', '456'), ('b', '789')])"
- assert h == http3.Headers([("a", "123"), ("b", "789"), ("a", "456")])
+ assert h == httpx.Headers([("a", "123"), ("b", "789"), ("a", "456")])
assert h != [("a", "123"), ("A", "456"), ("b", "789")]
- h = http3.Headers({"a": "123", "b": "789"})
+ h = httpx.Headers({"a": "123", "b": "789"})
assert h["A"] == "123"
assert h["B"] == "789"
assert h.raw == [(b"a", b"123"), (b"b", b"789")]
def test_header_mutations():
- h = http3.Headers()
+ h = httpx.Headers()
assert dict(h) == {}
h["a"] = "1"
assert dict(h) == {"a": "1"}
def test_copy_headers():
- headers = http3.Headers({"custom": "example"})
- headers_copy = http3.Headers(headers)
+ headers = httpx.Headers({"custom": "example"})
+ headers_copy = httpx.Headers(headers)
assert headers == headers_copy
def test_headers_insert_retains_ordering():
- headers = http3.Headers({"a": "a", "b": "b", "c": "c"})
+ headers = httpx.Headers({"a": "a", "b": "b", "c": "c"})
headers["b"] = "123"
assert list(headers.values()) == ["a", "123", "c"]
def test_headers_insert_appends_if_new():
- headers = http3.Headers({"a": "a", "b": "b", "c": "c"})
+ headers = httpx.Headers({"a": "a", "b": "b", "c": "c"})
headers["d"] = "123"
assert list(headers.values()) == ["a", "b", "c", "123"]
def test_headers_insert_removes_all_existing():
- headers = http3.Headers([("a", "123"), ("a", "456")])
+ headers = httpx.Headers([("a", "123"), ("a", "456")])
headers["a"] = "789"
assert dict(headers) == {"a": "789"}
def test_headers_delete_removes_all_existing():
- headers = http3.Headers([("a", "123"), ("a", "456")])
+ headers = httpx.Headers([("a", "123"), ("a", "456")])
del headers["a"]
assert dict(headers) == {}
"""
Headers should display with a dict repr by default.
"""
- headers = http3.Headers({"custom": "example"})
+ headers = httpx.Headers({"custom": "example"})
assert repr(headers) == "Headers({'custom': 'example'})"
"""
Headers should display an encoding in the repr if required.
"""
- headers = http3.Headers({b"custom": "example ☃".encode("utf-8")})
+ headers = httpx.Headers({b"custom": "example ☃".encode("utf-8")})
assert repr(headers) == "Headers({'custom': 'example ☃'}, encoding='utf-8')"
"""
Headers should display with a list repr if they include multiple identical keys.
"""
- headers = http3.Headers([("custom", "example 1"), ("custom", "example 2")])
+ headers = httpx.Headers([("custom", "example 1"), ("custom", "example 2")])
assert (
repr(headers) == "Headers([('custom', 'example 1'), ('custom', 'example 2')])"
)
Headers should decode as ascii by default.
"""
raw_headers = [(b"Custom", b"Example")]
- headers = http3.Headers(raw_headers)
+ headers = httpx.Headers(raw_headers)
assert dict(headers) == {"custom": "Example"}
assert headers.encoding == "ascii"
Headers containing non-ascii codepoints should default to decoding as utf-8.
"""
raw_headers = [(b"Custom", "Code point: ☃".encode("utf-8"))]
- headers = http3.Headers(raw_headers)
+ headers = httpx.Headers(raw_headers)
assert dict(headers) == {"custom": "Code point: ☃"}
assert headers.encoding == "utf-8"
Headers containing non-UTF-8 codepoints should default to decoding as iso-8859-1.
"""
raw_headers = [(b"Custom", "Code point: ÿ".encode("iso-8859-1"))]
- headers = http3.Headers(raw_headers)
+ headers = httpx.Headers(raw_headers)
assert dict(headers) == {"custom": "Code point: ÿ"}
assert headers.encoding == "iso-8859-1"
particular decoding.
"""
raw_headers = [(b"Custom", "Code point: ☃".encode("utf-8"))]
- headers = http3.Headers(raw_headers)
+ headers = httpx.Headers(raw_headers)
headers.encoding = "iso-8859-1"
assert dict(headers) == {"custom": "Code point: â\x98\x83"}
assert headers.encoding == "iso-8859-1"
"""
Most headers should split by commas for `getlist`, except 'Set-Cookie'.
"""
- h = http3.Headers([("set-cookie", "a, b"), ("set-cookie", "c")])
+ h = httpx.Headers([("set-cookie", "a, b"), ("set-cookie", "c")])
h.getlist("Set-Cookie") == ["a, b", "b"]
- h = http3.Headers([("vary", "a, b"), ("vary", "c")])
+ h = httpx.Headers([("vary", "a, b"), ("vary", "c")])
h.getlist("Vary") == ["a", "b", "c"]
-from http3 import QueryParams
+from httpx import QueryParams
def test_queryparams():
import pytest
-import http3
+import httpx
def test_request_repr():
- request = http3.Request("GET", "http://example.org")
+ request = httpx.Request("GET", "http://example.org")
assert repr(request) == "<Request('GET', 'http://example.org')>"
def test_no_content():
- request = http3.Request("GET", "http://example.org")
+ request = httpx.Request("GET", "http://example.org")
assert "Content-Length" not in request.headers
def test_content_length_header():
- request = http3.Request("POST", "http://example.org", data=b"test 123")
+ request = httpx.Request("POST", "http://example.org", data=b"test 123")
assert request.headers["Content-Length"] == "8"
def test_url_encoded_data():
- for RequestClass in (http3.Request, http3.AsyncRequest):
+ for RequestClass in (httpx.Request, httpx.AsyncRequest):
request = RequestClass("POST", "http://example.org", data={"test": "123"})
assert request.headers["Content-Type"] == "application/x-www-form-urlencoded"
assert request.content == b"test=123"
def test_json_encoded_data():
- for RequestClass in (http3.Request, http3.AsyncRequest):
+ for RequestClass in (httpx.Request, httpx.AsyncRequest):
request = RequestClass("POST", "http://example.org", json={"test": 123})
assert request.headers["Content-Type"] == "application/json"
assert request.content == b'{"test": 123}'
data = streaming_body(b"test 123")
- request = http3.Request("POST", "http://example.org", data=data)
+ request = httpx.Request("POST", "http://example.org", data=data)
assert "Content-Length" not in request.headers
assert request.headers["Transfer-Encoding"] == "chunked"
def test_override_host_header():
headers = {"host": "1.2.3.4:80"}
- request = http3.Request("GET", "http://example.org", headers=headers)
+ request = httpx.Request("GET", "http://example.org", headers=headers)
assert request.headers["Host"] == "1.2.3.4:80"
def test_override_accept_encoding_header():
headers = {"Accept-Encoding": "identity"}
- request = http3.Request("GET", "http://example.org", headers=headers)
+ request = httpx.Request("GET", "http://example.org", headers=headers)
assert request.headers["Accept-Encoding"] == "identity"
data = streaming_body(b"test 123")
headers = {"Content-Length": "8"}
- request = http3.Request("POST", "http://example.org", data=data, headers=headers)
+ request = httpx.Request("POST", "http://example.org", data=data, headers=headers)
assert request.headers["Content-Length"] == "8"
def test_url():
url = "http://example.org"
- request = http3.Request("GET", url)
+ request = httpx.Request("GET", url)
assert request.url.scheme == "http"
assert request.url.port == 80
assert request.url.full_path == "/"
url = "https://example.org/abc?foo=bar"
- request = http3.Request("GET", url)
+ request = httpx.Request("GET", url)
assert request.url.scheme == "https"
assert request.url.port == 443
assert request.url.full_path == "/abc?foo=bar"
def test_invalid_urls():
- with pytest.raises(http3.InvalidURL):
- http3.Request("GET", "example.org")
+ with pytest.raises(httpx.InvalidURL):
+ httpx.Request("GET", "example.org")
- with pytest.raises(http3.InvalidURL):
- http3.Request("GET", "http:///foo")
+ with pytest.raises(httpx.InvalidURL):
+ httpx.Request("GET", "http:///foo")
import pytest
-import http3
+import httpx
def streaming_body():
def test_response():
- response = http3.Response(200, content=b"Hello, world!")
+ response = httpx.Response(200, content=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
assert response.text == "Hello, world!"
def test_response_repr():
- response = http3.Response(200, content=b"Hello, world!")
+ response = httpx.Response(200, content=b"Hello, world!")
assert repr(response) == "<Response [200 OK]>"
"""
headers = {"Content-Type": "text-plain; charset=latin-1"}
content = "Latin 1: ÿ".encode("latin-1")
- response = http3.Response(200, content=content, headers=headers)
+ response = httpx.Response(200, content=content, headers=headers)
assert response.text == "Latin 1: ÿ"
assert response.encoding == "latin-1"
Autodetect encoding if there is no charset info in a Content-Type header.
"""
content = "おはようございます。".encode("EUC-JP")
- response = http3.Response(200, content=content)
+ response = httpx.Response(200, content=content)
assert response.text == "おはようございます。"
assert response.encoding == "EUC-JP"
"""
headers = {"Content-Type": "text-plain; charset=invalid-codec-name"}
content = "おはようございます。".encode("EUC-JP")
- response = http3.Response(200, content=content, headers=headers)
+ response = httpx.Response(200, content=content, headers=headers)
assert response.text == "おはようございます。"
assert response.encoding == "EUC-JP"
"""
content = b"Hello, world!"
headers = {"Content-Type": "text/plain"}
- response = http3.Response(200, content=content, headers=headers)
+ response = httpx.Response(200, content=content, headers=headers)
assert response.status_code == 200
assert response.encoding == "iso-8859-1"
assert response.text == "Hello, world!"
"""
Default to utf-8 if all else fails.
"""
- response = http3.Response(200, content=b"")
+ response = httpx.Response(200, content=b"")
assert response.text == ""
assert response.encoding == "utf-8"
Default to apparent encoding for non-text content-type headers.
"""
headers = {"Content-Type": "image/png"}
- response = http3.Response(200, content=b"xyz", headers=headers)
+ response = httpx.Response(200, content=b"xyz", headers=headers)
assert response.text == "xyz"
assert response.encoding == "ascii"
headers = {
"Content-Type": "text-plain; charset=utf-8"
} # Deliberately incorrect charset
- response = http3.Response(
+ response = httpx.Response(
200, content="Latin 1: ÿ".encode("latin-1"), headers=headers
)
response.encoding = "latin-1"
def test_response_force_encoding():
- response = http3.Response(200, content="Snowman: ☃".encode("utf-8"))
+ response = httpx.Response(200, content="Snowman: ☃".encode("utf-8"))
response.encoding = "iso-8859-1"
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_read_response():
- response = http3.Response(200, content=b"Hello, world!")
+ response = httpx.Response(200, content=b"Hello, world!")
assert response.status_code == 200
assert response.text == "Hello, world!"
def test_raw_interface():
- response = http3.Response(200, content=b"Hello, world!")
+ response = httpx.Response(200, content=b"Hello, world!")
raw = b""
for part in response.raw():
def test_stream_interface():
- response = http3.Response(200, content=b"Hello, world!")
+ response = httpx.Response(200, content=b"Hello, world!")
content = b""
for part in response.stream():
@pytest.mark.asyncio
async def test_async_stream_interface():
- response = http3.AsyncResponse(200, content=b"Hello, world!")
+ response = httpx.AsyncResponse(200, content=b"Hello, world!")
content = b""
async for part in response.stream():
def test_stream_interface_after_read():
- response = http3.Response(200, content=b"Hello, world!")
+ response = httpx.Response(200, content=b"Hello, world!")
response.read()
@pytest.mark.asyncio
async def test_async_stream_interface_after_read():
- response = http3.AsyncResponse(200, content=b"Hello, world!")
+ response = httpx.AsyncResponse(200, content=b"Hello, world!")
await response.read()
def test_streaming_response():
- response = http3.Response(200, content=streaming_body())
+ response = httpx.Response(200, content=streaming_body())
assert response.status_code == 200
assert not response.is_closed
@pytest.mark.asyncio
async def test_async_streaming_response():
- response = http3.AsyncResponse(200, content=async_streaming_body())
+ response = httpx.AsyncResponse(200, content=async_streaming_body())
assert response.status_code == 200
assert not response.is_closed
def test_cannot_read_after_stream_consumed():
- response = http3.Response(200, content=streaming_body())
+ response = httpx.Response(200, content=streaming_body())
content = b""
for part in response.stream():
content += part
- with pytest.raises(http3.StreamConsumed):
+ with pytest.raises(httpx.StreamConsumed):
response.read()
@pytest.mark.asyncio
async def test_async_cannot_read_after_stream_consumed():
- response = http3.AsyncResponse(200, content=async_streaming_body())
+ response = httpx.AsyncResponse(200, content=async_streaming_body())
content = b""
async for part in response.stream():
content += part
- with pytest.raises(http3.StreamConsumed):
+ with pytest.raises(httpx.StreamConsumed):
await response.read()
def test_cannot_read_after_response_closed():
- response = http3.Response(200, content=streaming_body())
+ response = httpx.Response(200, content=streaming_body())
response.close()
- with pytest.raises(http3.ResponseClosed):
+ with pytest.raises(httpx.ResponseClosed):
response.read()
@pytest.mark.asyncio
async def test_async_cannot_read_after_response_closed():
- response = http3.AsyncResponse(200, content=async_streaming_body())
+ response = httpx.AsyncResponse(200, content=async_streaming_body())
await response.close()
- with pytest.raises(http3.ResponseClosed):
+ with pytest.raises(httpx.ResponseClosed):
await response.read()
def test_unknown_status_code():
- response = http3.Response(600)
+ response = httpx.Response(600)
assert response.status_code == 600
assert response.reason_phrase == ""
assert response.text == ""
data = dict(greeting="hello", recipient="world")
content = json.dumps(data).encode("utf-16")
headers = {"Content-Type": "application/json, charset=utf-16"}
- response = http3.Response(200, content=content, headers=headers)
+ response = httpx.Response(200, content=content, headers=headers)
assert response.json() == data
data = dict(greeting="hello", recipient="world", amount=1)
content = json.dumps(data).encode("utf-16")
headers = {"Content-Type": "application/json, charset=utf-16"}
- response = http3.Response(200, content=content, headers=headers)
+ response = httpx.Response(200, content=content, headers=headers)
assert response.json(parse_int=str)["amount"] == "1"
data = dict(greeting="hello", recipient="world")
content = json.dumps(data).encode("utf-32-be")
headers = {"Content-Type": "application/json"}
- response = http3.Response(200, content=content, headers=headers)
+ response = httpx.Response(200, content=content, headers=headers)
assert response.json() == data
content = json.dumps(data).encode("utf-32-be")
headers = {"Content-Type": "application/json"}
# force incorrect guess from `guess_json_utf` to trigger error
- with mock.patch("http3.models.guess_json_utf", return_value="utf-32"):
- response = http3.Response(200, content=content, headers=headers)
+ with mock.patch("httpx.models.guess_json_utf", return_value="utf-32"):
+ response = httpx.Response(200, content=content, headers=headers)
with pytest.raises(json.JSONDecodeError):
response.json()
-from http3 import URL
+from httpx import URL
def test_idna_url():
import pytest
-import http3
+import httpx
def threadpool(func):
@threadpool
def test_get(server):
- response = http3.get("http://127.0.0.1:8000/")
+ response = httpx.get("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
assert response.text == "Hello, world!"
@threadpool
def test_post(server):
- response = http3.post("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = httpx.post("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
yield b", "
yield b"world!"
- response = http3.post("http://127.0.0.1:8000/", data=data())
+ response = httpx.post("http://127.0.0.1:8000/", data=data())
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_options(server):
- response = http3.options("http://127.0.0.1:8000/")
+ response = httpx.options("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_head(server):
- response = http3.head("http://127.0.0.1:8000/")
+ response = httpx.head("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_put(server):
- response = http3.put("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = httpx.put("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_patch(server):
- response = http3.patch("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = httpx.patch("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_delete(server):
- response = http3.delete("http://127.0.0.1:8000/")
+ response = httpx.delete("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_get_invalid_url(server):
- with pytest.raises(http3.InvalidURL):
- http3.get("invalid://example.org")
+ with pytest.raises(httpx.InvalidURL):
+ httpx.get("invalid://example.org")
import pytest
-import http3
+import httpx
async def hello_world(scope, receive, send):
def test_asgi():
- client = http3.Client(app=hello_world)
+ client = httpx.Client(app=hello_world)
response = client.get("http://www.example.org/")
assert response.status_code == 200
assert response.text == "Hello, World!"
def test_asgi_upload():
- client = http3.Client(app=echo_body)
+ client = httpx.Client(app=echo_body)
response = client.post("http://www.example.org/", data=b"example")
assert response.status_code == 200
assert response.text == "example"
def test_asgi_exc():
- client = http3.Client(app=raise_exc)
+ client = httpx.Client(app=raise_exc)
with pytest.raises(ValueError):
client.get("http://www.example.org/")
def test_asgi_exc_after_response():
- client = http3.Client(app=raise_exc_after_response)
+ client = httpx.Client(app=raise_exc_after_response)
with pytest.raises(ValueError):
client.get("http://www.example.org/")
import pytest
-import http3
+import httpx
@pytest.mark.asyncio
async def test_load_ssl_config():
- ssl_config = http3.SSLConfig()
+ ssl_config = httpx.SSLConfig()
context = await ssl_config.load_ssl_context()
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
assert context.check_hostname is True
@pytest.mark.asyncio
async def test_load_ssl_config_verify_non_existing_path():
- ssl_config = http3.SSLConfig(verify="/path/to/nowhere")
+ ssl_config = httpx.SSLConfig(verify="/path/to/nowhere")
with pytest.raises(IOError):
await ssl_config.load_ssl_context()
@pytest.mark.asyncio
async def test_load_ssl_config_verify_existing_file():
- ssl_config = http3.SSLConfig(verify=http3.config.DEFAULT_CA_BUNDLE_PATH)
+ ssl_config = httpx.SSLConfig(verify=httpx.config.DEFAULT_CA_BUNDLE_PATH)
context = await ssl_config.load_ssl_context()
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
assert context.check_hostname is True
@pytest.mark.asyncio
async def test_load_ssl_config_verify_directory():
- path = os.path.dirname(http3.config.DEFAULT_CA_BUNDLE_PATH)
- ssl_config = http3.SSLConfig(verify=path)
+ path = os.path.dirname(httpx.config.DEFAULT_CA_BUNDLE_PATH)
+ ssl_config = httpx.SSLConfig(verify=path)
context = await ssl_config.load_ssl_context()
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
assert context.check_hostname is True
@pytest.mark.asyncio
async def test_load_ssl_config_cert_and_key(cert_pem_file, cert_private_key_file):
- ssl_config = http3.SSLConfig(cert=(cert_pem_file, cert_private_key_file))
+ ssl_config = httpx.SSLConfig(cert=(cert_pem_file, cert_private_key_file))
context = await ssl_config.load_ssl_context()
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
assert context.check_hostname is True
async def test_load_ssl_config_cert_and_encrypted_key(
cert_pem_file, cert_encrypted_private_key_file, password
):
- ssl_config = http3.SSLConfig(
+ ssl_config = httpx.SSLConfig(
cert=(cert_pem_file, cert_encrypted_private_key_file, password)
)
context = await ssl_config.load_ssl_context()
async def test_load_ssl_config_cert_and_key_invalid_password(
cert_pem_file, cert_encrypted_private_key_file
):
- ssl_config = http3.SSLConfig(
+ ssl_config = httpx.SSLConfig(
cert=(cert_pem_file, cert_encrypted_private_key_file, "password1")
)
@pytest.mark.asyncio
async def test_load_ssl_config_cert_without_key_raises(cert_pem_file):
- ssl_config = http3.SSLConfig(cert=cert_pem_file)
+ ssl_config = httpx.SSLConfig(cert=cert_pem_file)
with pytest.raises(ssl.SSLError):
await ssl_config.load_ssl_context()
@pytest.mark.asyncio
async def test_load_ssl_config_no_verify():
- ssl_config = http3.SSLConfig(verify=False)
+ ssl_config = httpx.SSLConfig(verify=False)
context = await ssl_config.load_ssl_context()
assert context.verify_mode == ssl.VerifyMode.CERT_NONE
assert context.check_hostname is False
def test_ssl_repr():
- ssl = http3.SSLConfig(verify=False)
+ ssl = httpx.SSLConfig(verify=False)
assert repr(ssl) == "SSLConfig(cert=None, verify=False)"
def test_timeout_repr():
- timeout = http3.TimeoutConfig(timeout=5.0)
+ timeout = httpx.TimeoutConfig(timeout=5.0)
assert repr(timeout) == "TimeoutConfig(timeout=5.0)"
- timeout = http3.TimeoutConfig(read_timeout=5.0)
+ timeout = httpx.TimeoutConfig(read_timeout=5.0)
assert (
repr(timeout)
== "TimeoutConfig(connect_timeout=None, read_timeout=5.0, write_timeout=None)"
def test_limits_repr():
- limits = http3.PoolLimits(hard_limit=100)
+ limits = httpx.PoolLimits(hard_limit=100)
assert (
repr(limits) == "PoolLimits(soft_limit=None, hard_limit=100, pool_timeout=None)"
)
def test_ssl_eq():
- ssl = http3.SSLConfig(verify=False)
- assert ssl == http3.SSLConfig(verify=False)
+ ssl = httpx.SSLConfig(verify=False)
+ assert ssl == httpx.SSLConfig(verify=False)
def test_timeout_eq():
- timeout = http3.TimeoutConfig(timeout=5.0)
- assert timeout == http3.TimeoutConfig(timeout=5.0)
+ timeout = httpx.TimeoutConfig(timeout=5.0)
+ assert timeout == httpx.TimeoutConfig(timeout=5.0)
def test_limits_eq():
- limits = http3.PoolLimits(hard_limit=100)
- assert limits == http3.PoolLimits(hard_limit=100)
+ limits = httpx.PoolLimits(hard_limit=100)
+ assert limits == httpx.PoolLimits(hard_limit=100)
def test_timeout_from_tuple():
- timeout = http3.TimeoutConfig(timeout=(5.0, 5.0, 5.0))
- assert timeout == http3.TimeoutConfig(timeout=5.0)
+ timeout = httpx.TimeoutConfig(timeout=(5.0, 5.0, 5.0))
+ assert timeout == httpx.TimeoutConfig(timeout=5.0)
def test_timeout_from_config_instance():
- timeout = http3.TimeoutConfig(timeout=5.0)
- assert http3.TimeoutConfig(timeout) == http3.TimeoutConfig(timeout=5.0)
+ timeout = httpx.TimeoutConfig(timeout=5.0)
+ assert httpx.TimeoutConfig(timeout) == httpx.TimeoutConfig(timeout=5.0)
import brotli
import pytest
-import http3
+import httpx
def test_deflate():
compressed_body = compressor.compress(body) + compressor.flush()
headers = [(b"Content-Encoding", b"deflate")]
- response = http3.Response(200, headers=headers, content=compressed_body)
+ response = httpx.Response(200, headers=headers, content=compressed_body)
assert response.content == body
compressed_body = compressor.compress(body) + compressor.flush()
headers = [(b"Content-Encoding", b"gzip")]
- response = http3.Response(200, headers=headers, content=compressed_body)
+ response = httpx.Response(200, headers=headers, content=compressed_body)
assert response.content == body
compressed_body = brotli.compress(body)
headers = [(b"Content-Encoding", b"br")]
- response = http3.Response(200, headers=headers, content=compressed_body)
+ response = httpx.Response(200, headers=headers, content=compressed_body)
assert response.content == body
)
headers = [(b"Content-Encoding", b"deflate, gzip")]
- response = http3.Response(200, headers=headers, content=compressed_body)
+ response = httpx.Response(200, headers=headers, content=compressed_body)
assert response.content == body
compressed_body = brotli.compress(body)
headers = [(b"Content-Encoding", b"br, identity")]
- response = http3.Response(200, headers=headers, content=compressed_body)
+ response = httpx.Response(200, headers=headers, content=compressed_body)
assert response.content == body
headers = [(b"Content-Encoding", b"identity, br")]
- response = http3.Response(200, headers=headers, content=compressed_body)
+ response = httpx.Response(200, headers=headers, content=compressed_body)
assert response.content == body
yield compressor.flush()
headers = [(b"Content-Encoding", b"gzip")]
- response = http3.Response(200, headers=headers, content=compress(body))
+ response = httpx.Response(200, headers=headers, content=compress(body))
assert not hasattr(response, "body")
assert response.read() == body
headers = [(b"Content-Encoding", header_value)]
body = b"test 123"
compressed_body = brotli.compress(body)[3:]
- with pytest.raises(http3.exceptions.DecodingError):
- response = http3.Response(200, headers=headers, content=compressed_body)
+ with pytest.raises(httpx.exceptions.DecodingError):
+ response = httpx.Response(200, headers=headers, content=compressed_body)
response.content
import pytest
-from http3 import (
+from httpx import (
CertTypes,
Client,
Dispatcher,
-import http3
+import httpx
def test_status_code_as_int():
- assert http3.codes.NOT_FOUND == 404
- assert str(http3.codes.NOT_FOUND) == "404"
+ assert httpx.codes.NOT_FOUND == 404
+ assert str(httpx.codes.NOT_FOUND) == "404"
def test_lowercase_status_code():
- assert http3.codes.not_found == 404
+ assert httpx.codes.not_found == 404
def test_reason_phrase_for_status_code():
- assert http3.codes.get_reason_phrase(404) == "Not Found"
+ assert httpx.codes.get_reason_phrase(404) == "Not Found"
def test_reason_phrase_for_unknown_status_code():
- assert http3.codes.get_reason_phrase(499) == ""
+ assert httpx.codes.get_reason_phrase(499) == ""
import pytest
-from http3 import (
+from httpx import (
AsyncClient,
ConnectTimeout,
PoolLimits,
import pytest
-from http3.utils import guess_json_utf
+from httpx.utils import guess_json_utf
@pytest.mark.parametrize(
import pytest
-import http3
+import httpx
def hello_world(environ, start_response):
def test_wsgi():
- client = http3.Client(app=hello_world)
+ client = httpx.Client(app=hello_world)
response = client.get("http://www.example.org/")
assert response.status_code == 200
assert response.text == "Hello, World!"
def test_wsgi_upload():
- client = http3.Client(app=echo_body)
+ client = httpx.Client(app=echo_body)
response = client.post("http://www.example.org/", data=b"example")
assert response.status_code == 200
assert response.text == "example"
def test_wsgi_upload_with_response_stream():
- client = http3.Client(app=echo_body_with_response_stream)
+ client = httpx.Client(app=echo_body_with_response_stream)
response = client.post("http://www.example.org/", data=b"example")
assert response.status_code == 200
assert response.text == "example"
def test_wsgi_exc():
- client = http3.Client(app=raise_exc)
+ client = httpx.Client(app=raise_exc)
with pytest.raises(ValueError):
client.get("http://www.example.org/")
projects / thirdparty / httpx.git / commitdiff
