Compare commits

..

1 Commits

Author SHA1 Message Date
Tom Christie
54b8d24a6e Use system proxies 2024-11-05 14:55:32 +00:00
48 changed files with 868 additions and 1576 deletions

View File

@ -15,9 +15,9 @@ jobs:
steps:
- uses: "actions/checkout@v4"
- uses: "actions/setup-python@v6"
- uses: "actions/setup-python@v5"
with:
python-version: 3.9
python-version: 3.8
- name: "Install dependencies"
run: "scripts/install"
- name: "Build package & docs"

View File

@ -14,11 +14,11 @@ jobs:
strategy:
matrix:
python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
steps:
- uses: "actions/checkout@v4"
- uses: "actions/setup-python@v6"
- uses: "actions/setup-python@v5"
with:
python-version: "${{ matrix.python-version }}"
allow-prereleases: true

View File

@ -4,47 +4,20 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
## [UNRELEASED]
## [Unreleased]
### Removed
* Drop support for Python 3.8
### Added
* Expose `FunctionAuth` from the public API. (#3699)
## 0.28.1 (6th December, 2024)
* Fix SSL case where `verify=False` together with client side certificates.
## 0.28.0 (28th November, 2024)
Be aware that the default *JSON request bodies now use a more compact representation*. This is generally considered a prefered style, tho may require updates to test suites.
The 0.28 release includes a limited set of deprecations...
**Deprecations**:
We are working towards a simplified SSL configuration API.
*For users of the standard `verify=True` or `verify=False` cases, or `verify=<ssl_context>` case this should require no changes. The following cases have been deprecated...*
* The `verify` argument as a string argument is now deprecated and will raise warnings.
* The `cert` argument is now deprecated and will raise warnings.
Our revised [SSL documentation](docs/advanced/ssl.md) covers how to implement the same behaviour with a more constrained API.
**The following changes are also included**:
This release introduces an `httpx.SSLContext()` class and `ssl_context` parameter.
* Added `httpx.SSLContext` class and `ssl_context` parameter. (#3022, #3335)
* The `verify` and `cert` arguments have been deprecated and will now raise warnings. (#3022, #3335)
* The deprecated `proxies` argument has now been removed.
* The deprecated `app` argument has now been removed.
* JSON request bodies use a compact representation. (#3363)
* The `URL.raw` property has now been removed.
* Ensure JSON request bodies are compact. (#3363)
* Review URL percent escape sets, based on WHATWG spec. (#3371, #3373)
* Ensure `certifi` and `httpcore` are only imported if required. (#3377)
* Treat `socks5h` as a valid proxy scheme. (#3178)
* Cleanup `Request()` method signature in line with `client.request()` and `httpx.request()`. (#3378)
* Bugfix: When passing `params={}`, always strictly update rather than merge with an existing querystring. (#3364)
## 0.27.2 (27th August, 2024)
@ -632,7 +605,7 @@ See pull requests #1057, #1058.
* Added dedicated exception class `httpx.HTTPStatusError` for `.raise_for_status()` exceptions. (Pull #1072)
* Added `httpx.create_ssl_context()` helper function. (Pull #996)
* Support for proxy exclusions like `proxies={"https://www.example.com": None}`. (Pull #1099)
* Support for proxy exlcusions like `proxies={"https://www.example.com": None}`. (Pull #1099)
* Support `QueryParams(None)` and `client.params = None`. (Pull #1060)
### Changed
@ -860,7 +833,7 @@ We believe the API is now pretty much stable, and are aiming for a 1.0 release s
### Fixed
- Fix issue with concurrent connection acquisition. (Pull #700)
- Fix issue with concurrent connection acquiry. (Pull #700)
- Fix write error on closing HTTP/2 connections. (Pull #699)
## 0.10.0 (December 29th, 2019)
@ -1109,7 +1082,7 @@ importing modules within the package.
## 0.6.7 (July 8, 2019)
- Check for connection aliveness on re-acquisition (Pull #111)
- Check for connection aliveness on re-acquiry (Pull #111)
## 0.6.6 (July 3, 2019)

View File

@ -13,7 +13,9 @@
</a>
</p>
HTTPX is a fully featured HTTP client library for Python 3. It includes **an integrated command line client**, has support for both **HTTP/1.1 and HTTP/2**, and provides both **sync and async APIs**.
HTTPX is a fully featured HTTP client library for Python 3. It includes **an integrated
command line client**, has support for both **HTTP/1.1 and HTTP/2**, and provides both **sync
and async APIs**.
---
@ -101,7 +103,7 @@ Or, to include the optional HTTP/2 support, use:
$ pip install httpx[http2]
```
HTTPX requires Python 3.9+.
HTTPX requires Python 3.8+.
## Documentation

View File

@ -270,9 +270,8 @@ multipart file encoding is available by passing a dictionary with the
name of the payloads as keys and either tuple of elements or a file-like object or a string as values.
```pycon
>>> with open('report.xls', 'rb') as report_file:
... files = {'upload-file': ('report.xls', report_file, 'application/vnd.ms-excel')}
... r = httpx.post("https://httpbin.org/post", files=files)
>>> files = {'upload-file': ('report.xls', open('report.xls', 'rb'), 'application/vnd.ms-excel')}
>>> r = httpx.post("https://httpbin.org/post", files=files)
>>> print(r.text)
{
...
@ -319,10 +318,7 @@ To do that, pass a list of `(field, <file>)` items instead of a dictionary, allo
For instance this request sends 2 files, `foo.png` and `bar.png` in one request on the `images` form field:
```pycon
>>> with open('foo.png', 'rb') as foo_file, open('bar.png', 'rb') as bar_file:
... files = [
... ('images', ('foo.png', foo_file, 'image/png')),
... ('images', ('bar.png', bar_file, 'image/png')),
... ]
... r = httpx.post("https://httpbin.org/post", files=files)
>>> files = [('images', ('foo.png', open('foo.png', 'rb'), 'image/png')),
('images', ('bar.png', open('bar.png', 'rb'), 'image/png'))]
>>> r = httpx.post("https://httpbin.org/post", files=files)
```

View File

@ -9,81 +9,191 @@ By default httpx will verify HTTPS connections, and raise an error for invalid S
httpx.ConnectError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: certificate has expired (_ssl.c:997)
```
You can disable SSL verification completely and allow insecure requests...
Verification is configured through [the SSL Context API](https://docs.python.org/3/library/ssl.html#ssl-contexts).
```pycon
>>> httpx.get("https://expired.badssl.com/", verify=False)
>>> context = httpx.SSLContext()
>>> context
<SSLContext(verify=True)>
>>> httpx.get("https://www.example.com", ssl_context=context)
httpx.ConnectError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: certificate has expired (_ssl.c:997)
```
You can use this to disable verification completely and allow insecure requests...
```pycon
>>> context = httpx.SSLContext(verify=False)
>>> context
<SSLContext(verify=False)>
>>> httpx.get("https://expired.badssl.com/", ssl_context=context)
<Response [200 OK]>
```
### Configuring client instances
If you're using a `Client()` instance you should pass any `verify=<...>` configuration when instantiating the client.
If you're using a `Client()` instance you should pass any SSL context when instantiating the client.
By default the [certifi CA bundle](https://certifiio.readthedocs.io/en/latest/) is used for SSL verification.
For more complex configurations you can pass an [SSL Context](https://docs.python.org/3/library/ssl.html) instance...
```python
import certifi
import httpx
import ssl
# This SSL context is equivalent to the default `verify=True`.
ctx = ssl.create_default_context(cafile=certifi.where())
client = httpx.Client(verify=ctx)
```pycon
>>> context = httpx.SSLContext()
>>> client = httpx.Client(ssl_context=context)
```
Using [the `truststore` package](https://truststore.readthedocs.io/) to support system certificate stores...
The `client.get(...)` method and other request methods on a `Client` instance *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 than 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.
### Configuring certificate stores
By default, HTTPX uses the CA bundle provided by [Certifi](https://pypi.org/project/certifi/).
You can load additional certificate verification using the [`.load_verify_locations()`](https://docs.python.org/3/library/ssl.html#ssl.SSLContext.load_verify_locations) API:
```pycon
>>> context = httpx.SSLContext()
>>> context.load_verify_locations(cafile="path/to/certs.pem")
>>> client = httpx.Client(ssl_context=context)
>>> client.get("https://www.example.com")
<Response [200 OK]>
```
Or by providing an certificate directory:
```pycon
>>> context = httpx.SSLContext()
>>> context.load_verify_locations(capath="path/to/certs")
>>> client = httpx.Client(ssl_context=context)
>>> client.get("https://www.example.com")
<Response [200 OK]>
```
### Client side certificates
You can also specify a local cert to use as a client-side certificate, using the [`.load_cert_chain()`](https://docs.python.org/3/library/ssl.html#ssl.SSLContext.load_cert_chain) API:
```pycon
>>> context = httpx.SSLContext()
>>> context.load_cert_chain(certfile="path/to/client.pem")
>>> httpx.get("https://example.org", ssl_context=ssl_context)
<Response [200 OK]>
```
Or including a keyfile...
```pycon
>>> context = httpx.SSLContext()
>>> context.load_cert_chain(
certfile="path/to/client.pem",
keyfile="path/to/client.key"
)
>>> httpx.get("https://example.org", ssl_context=context)
<Response [200 OK]>
```
Or including a keyfile and password...
```pycon
>>> context = httpx.SSLContext(cert=cert)
>>> context = httpx.SSLContext()
>>> context.load_cert_chain(
certfile="path/to/client.pem",
keyfile="path/to/client.key"
password="password"
)
>>> httpx.get("https://example.org", ssl_context=context)
<Response [200 OK]>
```
### Using alternate SSL contexts
You can also use an alternate `ssl.SSLContext` instances.
For example, [using the `truststore` package](https://truststore.readthedocs.io/)...
```python
import ssl
import truststore
import httpx
# Use system certificate stores.
ctx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
client = httpx.Client(verify=ctx)
ssl_context = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
client = httpx.Client(ssl_context=ssl_context)
```
Loding an alternative certificate verification store using [the standard SSL context API](https://docs.python.org/3/library/ssl.html)...
Or working [directly with Python's standard library](https://docs.python.org/3/library/ssl.html)...
```python
import httpx
import ssl
import httpx
# Use an explicitly configured certificate store.
ctx = ssl.create_default_context(cafile="path/to/certs.pem") # Either cafile or capath.
client = httpx.Client(verify=ctx)
```
### Client side certificates
Client side certificates allow a remote server to verify the client. They tend to be used within private organizations to authenticate requests to remote servers.
You can specify client-side certificates, using the [`.load_cert_chain()`](https://docs.python.org/3/library/ssl.html#ssl.SSLContext.load_cert_chain) API...
```python
ctx = ssl.create_default_context()
ctx.load_cert_chain(certfile="path/to/client.pem") # Optionally also keyfile or password.
client = httpx.Client(verify=ctx)
ssl_context = ssl.create_default_context()
client = httpx.Client(ssl_context=ssl_context)
```
### Working with `SSL_CERT_FILE` and `SSL_CERT_DIR`
`httpx` does respect the `SSL_CERT_FILE` and `SSL_CERT_DIR` environment variables by default. For details, refer to [the section on the environment variables page](../environment_variables.md#ssl_cert_file).
Unlike `requests`, the `httpx` package does not automatically pull in [the environment variables `SSL_CERT_FILE` or `SSL_CERT_DIR`](https://www.openssl.org/docs/manmaster/man3/SSL_CTX_set_default_verify_paths.html). If you want to use these they need to be enabled explicitly.
For example...
```python
context = httpx.SSLContext()
# Use `SSL_CERT_FILE` or `SSL_CERT_DIR` if configured.
if os.environ.get("SSL_CERT_FILE") or os.environ.get("SSL_CERT_DIR"):
context.load_verify_locations(
cafile=os.environ.get("SSL_CERT_FILE"),
capath=os.environ.get("SSL_CERT_DIR"),
)
```
## `SSLKEYLOGFILE`
Valid values: a filename
If this environment variable is set, TLS keys will be appended to the specified file, creating it if it doesn't exist, whenever key material is generated or received. The keylog file is designed for debugging purposes only.
Support for `SSLKEYLOGFILE` requires Python 3.8 and OpenSSL 1.1.1 or newer.
Example:
```python
# test_script.py
import httpx
with httpx.Client() as client:
r = client.get("https://google.com")
```
```console
SSLKEYLOGFILE=test.log python test_script.py
cat test.log
# TLS secrets log file, generated by OpenSSL / Python
SERVER_HANDSHAKE_TRAFFIC_SECRET XXXX
EXPORTER_SECRET XXXX
SERVER_TRAFFIC_SECRET_0 XXXX
CLIENT_HANDSHAKE_TRAFFIC_SECRET XXXX
CLIENT_TRAFFIC_SECRET_0 XXXX
SERVER_HANDSHAKE_TRAFFIC_SECRET XXXX
EXPORTER_SECRET XXXX
SERVER_TRAFFIC_SECRET_0 XXXX
CLIENT_HANDSHAKE_TRAFFIC_SECRET XXXX
CLIENT_TRAFFIC_SECRET_0 XXXX
```
### Making HTTPS requests to a local server
When making requests to local servers, such as a development server running on `localhost`, you will typically be using unencrypted HTTP connections.
If you do need to make HTTPS connections to a local server, for example to test an HTTPS-only service, you will need to create and use your own certificates. Here's one way to do it...
If you do need to make HTTPS connections to a local server, for example to test an HTTPS-only service, you will need to create and use your own certificates. Here's one way to do it:
1. Use [trustme](https://github.com/python-trio/trustme) to generate a pair of server key/cert files, and a client cert file.
2. Pass the server key/cert files when starting your local server. (This depends on the particular web server you're using. For example, [Uvicorn](https://www.uvicorn.org) provides the `--ssl-keyfile` and `--ssl-certfile` options.)
3. Configure `httpx` to use the certificates stored in `client.pem`.
3. Tell HTTPX to use the certificates stored in `client.pem`:
```python
ctx = ssl.create_default_context(cafile="client.pem")
client = httpx.Client(verify=ctx)
```pycon
>>> import httpx
>>> context = httpx.SSLContext()
>>> context.load_verify_locations(cafile="/tmp/client.pem")
>>> r = httpx.get("https://localhost:8000", ssl_context=context)
>>> r
Response <200 OK>
```

View File

@ -159,18 +159,3 @@ what gets sent over the wire.*
* `def delete(name, [domain], [path])`
* `def clear([domain], [path])`
* *Standard mutable mapping interface*
## `Proxy`
*A configuration of the proxy server.*
```pycon
>>> proxy = Proxy("http://proxy.example.com:8030")
>>> client = Client(proxy=proxy)
```
* `def __init__(url, [ssl_context], [auth], [headers])`
* `.url` - **URL**
* `.auth` - **tuple[str, str]**
* `.headers` - **Headers**
* `.ssl_context` - **SSLContext**

View File

@ -23,7 +23,7 @@ To make asynchronous requests, you'll need an `AsyncClient`.
```
!!! tip
Use [IPython](https://ipython.readthedocs.io/en/stable/) or Python 3.9+ with `python -m asyncio` to try this code interactively, as they support executing `async`/`await` expressions in the console.
Use [IPython](https://ipython.readthedocs.io/en/stable/) or Python 3.8+ with `python -m asyncio` to try this code interactively, as they support executing `async`/`await` expressions in the console.
## API Differences

View File

@ -226,7 +226,3 @@ For both query params (`params=`) and form data (`data=`), `requests` supports s
In HTTPX, event hooks may access properties of requests and responses, but event hook callbacks cannot mutate the original request/response.
If you are looking for more control, consider checking out [Custom Transports](advanced/transports.md#custom-transports).
## Exceptions and Errors
`requests` exception hierarchy is slightly different to the `httpx` exception hierarchy. `requests` exposes a top level `RequestException`, where as `httpx` exposes a top level `HTTPError`. see the exceptions exposes in requests [here](https://requests.readthedocs.io/en/latest/_modules/requests/exceptions/). See the `httpx` error hierarchy [here](https://www.python-httpx.org/exceptions/).

View File

@ -210,9 +210,12 @@ configure HTTPX as described in the
the [SSL certificates section](https://www.python-httpx.org/advanced/ssl/),
this is where our previously generated `client.pem` comes in:
```python
ctx = ssl.create_default_context(cafile="/path/to/client.pem")
client = httpx.Client(proxy="http://127.0.0.1:8080/", verify=ctx)
```
import httpx
with httpx.Client(proxy="http://127.0.0.1:8080/", verify="/path/to/client.pem") as client:
response = client.get("https://example.org")
print(response.status_code) # should print 200
```
Note, however, that HTTPS requests will only succeed to the host specified

View File

@ -51,29 +51,3 @@ python -c "import httpx; httpx.get('http://example.com')"
python -c "import httpx; httpx.get('http://127.0.0.1:5000/my-api')"
python -c "import httpx; httpx.get('https://www.python-httpx.org')"
```
## `SSL_CERT_FILE`
Valid values: a filename
If this environment variable is set then HTTPX will load
CA certificate from the specified file instead of the default
location.
Example:
```console
SSL_CERT_FILE=/path/to/ca-certs/ca-bundle.crt python -c "import httpx; httpx.get('https://example.com')"
```
## `SSL_CERT_DIR`
Valid values: a directory following an [OpenSSL specific layout](https://www.openssl.org/docs/manmaster/man3/SSL_CTX_load_verify_locations.html).
If this environment variable is set and the directory follows an [OpenSSL specific layout](https://www.openssl.org/docs/manmaster/man3/SSL_CTX_load_verify_locations.html) (ie. you ran `c_rehash`) then HTTPX will load CA certificates from this directory instead of the default location.
Example:
```console
SSL_CERT_DIR=/path/to/ca-certs/ python -c "import httpx; httpx.get('https://example.com')"
```

Binary file not shown.

Before

Width:  |  Height:  |  Size: 113 KiB

After

Width:  |  Height:  |  Size: 1.9 MiB

View File

@ -145,6 +145,6 @@ To include the optional brotli and zstandard decoders support, use:
$ pip install httpx[brotli,zstd]
```
HTTPX requires Python 3.9+
HTTPX requires Python 3.8+
[sync-support]: https://github.com/encode/httpx/issues/572

View File

@ -20,6 +20,8 @@ httpx.get("https://www.example.com")
Will send debug level output to the console, or wherever `stdout` is directed too...
```
DEBUG [2024-09-28 17:27:40] httpx - load_ssl_context verify=True cert=None
DEBUG [2024-09-28 17:27:40] httpx - load_verify_locations cafile='/Users/karenpetrosyan/oss/karhttpx/.venv/lib/python3.9/site-packages/certifi/cacert.pem'
DEBUG [2024-09-28 17:27:40] httpcore.connection - connect_tcp.started host='www.example.com' port=443 local_address=None timeout=5.0 socket_options=None
DEBUG [2024-09-28 17:27:41] httpcore.connection - connect_tcp.complete return_value=<httpcore._backends.sync.SyncStream object at 0x101f1e8e0>
DEBUG [2024-09-28 17:27:41] httpcore.connection - start_tls.started ssl_context=SSLContext(verify=True) server_hostname='www.example.com' timeout=5.0
@ -78,4 +80,4 @@ logging.config.dictConfig(LOGGING_CONFIG)
httpx.get('https://www.example.com')
```
The exact formatting of the debug logging may be subject to change across different versions of `httpx` and `httpcore`. If you need to rely on a particular format it is recommended that you pin installation of these packages to fixed versions.
The exact formatting of the debug logging may be subject to change across different versions of `httpx` and `httpcore`. If you need to rely on a particular format it is recommended that you pin installation of these packages to fixed versions.

View File

@ -174,9 +174,8 @@ Form encoded data can also include multiple values from a given key.
You can also upload files, using HTTP multipart encoding:
```pycon
>>> with open('report.xls', 'rb') as report_file:
... files = {'upload-file': report_file}
... r = httpx.post("https://httpbin.org/post", files=files)
>>> files = {'upload-file': open('report.xls', 'rb')}
>>> r = httpx.post("https://httpbin.org/post", files=files)
>>> print(r.text)
{
...
@ -191,9 +190,8 @@ You can also explicitly set the filename and content type, by using a tuple
of items for the file value:
```pycon
>>> with open('report.xls', 'rb') as report_file:
... files = {'upload-file': ('report.xls', report_file, 'application/vnd.ms-excel')}
... r = httpx.post("https://httpbin.org/post", files=files)
>>> files = {'upload-file': ('report.xls', open('report.xls', 'rb'), 'application/vnd.ms-excel')}
>>> r = httpx.post("https://httpbin.org/post", files=files)
>>> print(r.text)
{
...
@ -208,9 +206,8 @@ If you need to include non-file data fields in the multipart form, use the `data
```pycon
>>> data = {'message': 'Hello, world!'}
>>> with open('report.xls', 'rb') as report_file:
... files = {'file': report_file}
... r = httpx.post("https://httpbin.org/post", data=data, files=files)
>>> files = {'file': open('report.xls', 'rb')}
>>> r = httpx.post("https://httpbin.org/post", data=data, files=files)
>>> print(r.text)
{
...

View File

@ -2,83 +2,31 @@
As HTTPX usage grows, there is an expanding community of developers building tools and libraries that integrate with HTTPX, or depend on HTTPX. Here are some of them.
<!-- NOTE: Entries are alphabetised. -->
## Plugins
### Hishel
[GitHub](https://github.com/karpetrosyan/hishel) - [Documentation](https://hishel.com/)
An elegant HTTP Cache implementation for HTTPX and HTTP Core.
### HTTPX-Auth
[GitHub](https://github.com/Colin-b/httpx_auth) - [Documentation](https://colin-b.github.io/httpx_auth/)
Provides authentication classes to be used with HTTPX's [authentication parameter](advanced/authentication.md#customizing-authentication).
### httpx-caching
[Github](https://github.com/johtso/httpx-caching)
This package adds caching functionality to HTTPX
### httpx-secure
[GitHub](https://github.com/Zaczero/httpx-secure)
Drop-in SSRF protection for httpx with DNS caching and custom validation support.
### httpx-socks
[GitHub](https://github.com/romis2012/httpx-socks)
Proxy (HTTP, SOCKS) transports for httpx.
### httpx-sse
[GitHub](https://github.com/florimondmanca/httpx-sse)
Allows consuming Server-Sent Events (SSE) with HTTPX.
### httpx-retries
[GitHub](https://github.com/will-ockmore/httpx-retries) - [Documentation](https://will-ockmore.github.io/httpx-retries/)
A retry layer for HTTPX.
### httpx-ws
[GitHub](https://github.com/frankie567/httpx-ws) - [Documentation](https://frankie567.github.io/httpx-ws/)
WebSocket support for HTTPX.
### pytest-HTTPX
### httpx-socks
[GitHub](https://github.com/Colin-b/pytest_httpx) - [Documentation](https://colin-b.github.io/pytest_httpx/)
[GitHub](https://github.com/romis2012/httpx-socks)
Provides a [pytest](https://docs.pytest.org/en/latest/) fixture to mock HTTPX within test cases.
Proxy (HTTP, SOCKS) transports for httpx.
### RESPX
### Hishel
[GitHub](https://github.com/lundberg/respx) - [Documentation](https://lundberg.github.io/respx/)
[GitHub](https://github.com/karpetrosyan/hishel) - [Documentation](https://hishel.com/)
A utility for mocking out HTTPX.
### rpc.py
[Github](https://github.com/abersheeran/rpc.py) - [Documentation](https://github.com/abersheeran/rpc.py#rpcpy)
A fast and powerful RPC framework based on ASGI/WSGI. Use HTTPX as the client of the RPC service.
## Libraries with HTTPX support
An elegant HTTP Cache implementation for HTTPX and HTTP Core.
### Authlib
[GitHub](https://github.com/lepture/authlib) - [Documentation](https://docs.authlib.org/en/latest/)
A python library for building OAuth and OpenID Connect clients and servers. Includes an [OAuth HTTPX client](https://docs.authlib.org/en/latest/client/httpx.html).
The ultimate Python library in building OAuth and OpenID Connect clients and servers. Includes an [OAuth HTTPX client](https://docs.authlib.org/en/latest/client/httpx.html).
### Gidgethub
@ -86,20 +34,58 @@ A python library for building OAuth and OpenID Connect clients and servers. Incl
An asynchronous GitHub API library. Includes [HTTPX support](https://gidgethub.readthedocs.io/en/latest/httpx.html).
### httpdbg
### HTTPX-Auth
[GitHub](https://github.com/cle-b/httpdbg) - [Documentation](https://httpdbg.readthedocs.io/)
[GitHub](https://github.com/Colin-b/httpx_auth) - [Documentation](https://colin-b.github.io/httpx_auth/)
A tool for python developers to easily debug the HTTP(S) client requests in a python program.
Provides authentication classes to be used with HTTPX [authentication parameter](advanced/authentication.md#customizing-authentication).
### pytest-HTTPX
[GitHub](https://github.com/Colin-b/pytest_httpx) - [Documentation](https://colin-b.github.io/pytest_httpx/)
Provides `httpx_mock` [pytest](https://docs.pytest.org/en/latest/) fixture to mock HTTPX within test cases.
### RESPX
[GitHub](https://github.com/lundberg/respx) - [Documentation](https://lundberg.github.io/respx/)
A utility for mocking out the Python HTTPX library.
### rpc.py
[Github](https://github.com/abersheeran/rpc.py) - [Documentation](https://github.com/abersheeran/rpc.py#rpcpy)
An fast and powerful RPC framework based on ASGI/WSGI. Use HTTPX as the client of the RPC service.
### VCR.py
[GitHub](https://github.com/kevin1024/vcrpy) - [Documentation](https://vcrpy.readthedocs.io/)
Record and repeat requests.
A utility for record and repeat an http request.
### httpx-caching
[Github](https://github.com/johtso/httpx-caching)
This package adds caching functionality to HTTPX
### httpx-sse
[GitHub](https://github.com/florimondmanca/httpx-sse)
Allows consuming Server-Sent Events (SSE) with HTTPX.
### robox
[Github](https://github.com/danclaudiupop/robox)
A library for scraping the web built on top of HTTPX.
## Gists
<!-- NOTE: this list is in alphabetical order. -->
### urllib3-transport
[GitHub](https://gist.github.com/florimondmanca/d56764d78d748eb9f73165da388e546e)

View File

@ -50,7 +50,6 @@ __all__ = [
"DecodingError",
"delete",
"DigestAuth",
"FunctionAuth",
"get",
"head",
"Headers",
@ -82,6 +81,7 @@ __all__ = [
"RequestNotRead",
"Response",
"ResponseNotRead",
"SSLContext",
"stream",
"StreamClosed",
"StreamConsumed",

View File

@ -1,3 +1,3 @@
__title__ = "httpx"
__description__ = "A next generation HTTP client, for Python 3."
__version__ = "0.28.1"
__version__ = "0.27.2"

View File

@ -1,5 +1,6 @@
from __future__ import annotations
import ssl
import typing
from contextlib import contextmanager
@ -19,10 +20,6 @@ from ._types import (
)
from ._urls import URL
if typing.TYPE_CHECKING:
import ssl # pragma: no cover
__all__ = [
"delete",
"get",
@ -51,8 +48,11 @@ def request(
proxy: ProxyTypes | None = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
follow_redirects: bool = False,
verify: ssl.SSLContext | str | bool = True,
ssl_context: ssl.SSLContext | None = None,
trust_env: bool = True,
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> Response:
"""
Sends an HTTP request.
@ -82,9 +82,8 @@ def request(
* **timeout** - *(optional)* The timeout configuration to use when sending
the request.
* **follow_redirects** - *(optional)* Enables or disables HTTP redirects.
* **verify** - *(optional)* Either `True` to use an SSL context with the
default CA bundle, `False` to disable verification, or an instance of
`ssl.SSLContext` to use a custom context.
* **ssl_context** - *(optional)* An SSL certificate used by the requested host
to authenticate the client.
* **trust_env** - *(optional)* Enables or disables usage of environment
variables for configuration.
@ -102,9 +101,11 @@ def request(
with Client(
cookies=cookies,
proxy=proxy,
verify=verify,
ssl_context=ssl_context,
timeout=timeout,
trust_env=trust_env,
verify=verify,
cert=cert,
) as client:
return client.request(
method=method,
@ -136,8 +137,11 @@ def stream(
proxy: ProxyTypes | None = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
follow_redirects: bool = False,
verify: ssl.SSLContext | str | bool = True,
ssl_context: ssl.SSLContext | None = None,
trust_env: bool = True,
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> typing.Iterator[Response]:
"""
Alternative to `httpx.request()` that streams the response body
@ -152,9 +156,11 @@ def stream(
with Client(
cookies=cookies,
proxy=proxy,
verify=verify,
ssl_context=ssl_context,
timeout=timeout,
trust_env=trust_env,
verify=verify,
cert=cert,
) as client:
with client.stream(
method=method,
@ -180,9 +186,12 @@ def get(
auth: AuthTypes | None = None,
proxy: ProxyTypes | None = None,
follow_redirects: bool = False,
verify: ssl.SSLContext | str | bool = True,
ssl_context: ssl.SSLContext | None = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
trust_env: bool = True,
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> Response:
"""
Sends a `GET` request.
@ -201,9 +210,11 @@ def get(
auth=auth,
proxy=proxy,
follow_redirects=follow_redirects,
verify=verify,
ssl_context=ssl_context,
timeout=timeout,
trust_env=trust_env,
verify=verify,
cert=cert,
)
@ -216,9 +227,12 @@ def options(
auth: AuthTypes | None = None,
proxy: ProxyTypes | None = None,
follow_redirects: bool = False,
verify: ssl.SSLContext | str | bool = True,
ssl_context: ssl.SSLContext | None = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
trust_env: bool = True,
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> Response:
"""
Sends an `OPTIONS` request.
@ -237,9 +251,11 @@ def options(
auth=auth,
proxy=proxy,
follow_redirects=follow_redirects,
verify=verify,
ssl_context=ssl_context,
timeout=timeout,
trust_env=trust_env,
verify=verify,
cert=cert,
)
@ -252,9 +268,12 @@ def head(
auth: AuthTypes | None = None,
proxy: ProxyTypes | None = None,
follow_redirects: bool = False,
verify: ssl.SSLContext | str | bool = True,
ssl_context: ssl.SSLContext | None = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
trust_env: bool = True,
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> Response:
"""
Sends a `HEAD` request.
@ -273,9 +292,11 @@ def head(
auth=auth,
proxy=proxy,
follow_redirects=follow_redirects,
verify=verify,
ssl_context=ssl_context,
timeout=timeout,
trust_env=trust_env,
verify=verify,
cert=cert,
)
@ -292,9 +313,12 @@ def post(
auth: AuthTypes | None = None,
proxy: ProxyTypes | None = None,
follow_redirects: bool = False,
verify: ssl.SSLContext | str | bool = True,
ssl_context: ssl.SSLContext | None = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
trust_env: bool = True,
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> Response:
"""
Sends a `POST` request.
@ -314,9 +338,11 @@ def post(
auth=auth,
proxy=proxy,
follow_redirects=follow_redirects,
verify=verify,
ssl_context=ssl_context,
timeout=timeout,
trust_env=trust_env,
verify=verify,
cert=cert,
)
@ -333,9 +359,12 @@ def put(
auth: AuthTypes | None = None,
proxy: ProxyTypes | None = None,
follow_redirects: bool = False,
verify: ssl.SSLContext | str | bool = True,
ssl_context: ssl.SSLContext | None = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
trust_env: bool = True,
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> Response:
"""
Sends a `PUT` request.
@ -355,9 +384,11 @@ def put(
auth=auth,
proxy=proxy,
follow_redirects=follow_redirects,
verify=verify,
ssl_context=ssl_context,
timeout=timeout,
trust_env=trust_env,
verify=verify,
cert=cert,
)
@ -374,9 +405,12 @@ def patch(
auth: AuthTypes | None = None,
proxy: ProxyTypes | None = None,
follow_redirects: bool = False,
verify: ssl.SSLContext | str | bool = True,
ssl_context: ssl.SSLContext | None = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
trust_env: bool = True,
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> Response:
"""
Sends a `PATCH` request.
@ -396,9 +430,11 @@ def patch(
auth=auth,
proxy=proxy,
follow_redirects=follow_redirects,
verify=verify,
ssl_context=ssl_context,
timeout=timeout,
trust_env=trust_env,
verify=verify,
cert=cert,
)
@ -412,8 +448,11 @@ def delete(
proxy: ProxyTypes | None = None,
follow_redirects: bool = False,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
verify: ssl.SSLContext | str | bool = True,
ssl_context: ssl.SSLContext | None = None,
trust_env: bool = True,
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> Response:
"""
Sends a `DELETE` request.
@ -432,7 +471,9 @@ def delete(
auth=auth,
proxy=proxy,
follow_redirects=follow_redirects,
verify=verify,
ssl_context=ssl_context,
timeout=timeout,
trust_env=trust_env,
verify=verify,
cert=cert,
)

View File

@ -16,7 +16,7 @@ if typing.TYPE_CHECKING: # pragma: no cover
from hashlib import _Hash
__all__ = ["Auth", "BasicAuth", "DigestAuth", "FunctionAuth", "NetRCAuth"]
__all__ = ["Auth", "BasicAuth", "DigestAuth", "NetRCAuth"]
class Auth:

View File

@ -3,6 +3,7 @@ from __future__ import annotations
import datetime
import enum
import logging
import ssl
import time
import typing
import warnings
@ -16,7 +17,6 @@ from ._config import (
DEFAULT_MAX_REDIRECTS,
DEFAULT_TIMEOUT_CONFIG,
Limits,
Proxy,
Timeout,
)
from ._decoders import SUPPORTED_DECODERS
@ -33,7 +33,6 @@ from ._transports.default import AsyncHTTPTransport, HTTPTransport
from ._types import (
AsyncByteStream,
AuthTypes,
CertTypes,
CookieTypes,
HeaderTypes,
ProxyTypes,
@ -46,10 +45,10 @@ from ._types import (
TimeoutTypes,
)
from ._urls import URL, QueryParams
from ._utils import URLPattern, get_environment_proxies
if typing.TYPE_CHECKING:
import ssl # pragma: no cover
from ._utils import (
is_https_redirect,
same_origin,
)
__all__ = ["USE_CLIENT_DEFAULT", "AsyncClient", "Client"]
@ -59,38 +58,6 @@ T = typing.TypeVar("T", bound="Client")
U = typing.TypeVar("U", bound="AsyncClient")
def _is_https_redirect(url: URL, location: URL) -> bool:
"""
Return 'True' if 'location' is a HTTPS upgrade of 'url'
"""
if url.host != location.host:
return False
return (
url.scheme == "http"
and _port_or_default(url) == 80
and location.scheme == "https"
and _port_or_default(location) == 443
)
def _port_or_default(url: URL) -> int | None:
if url.port is not None:
return url.port
return {"http": 80, "https": 443}.get(url.scheme)
def _same_origin(url: URL, other: URL) -> bool:
"""
Return 'True' if the given URLs share the same origin.
"""
return (
url.scheme == other.scheme
and url.host == other.host
and _port_or_default(url) == _port_or_default(other)
)
class UseClientDefault:
"""
For some parameters such as `auth=...` and `timeout=...` we need to be able
@ -236,20 +203,6 @@ class BaseClient:
return url
return url.copy_with(raw_path=url.raw_path + b"/")
def _get_proxy_map(
self, proxy: ProxyTypes | None, allow_env_proxies: bool
) -> dict[str, Proxy | None]:
if proxy is None:
if allow_env_proxies:
return {
key: None if url is None else Proxy(url=url)
for key, url in get_environment_proxies().items()
}
return {}
else:
proxy = Proxy(url=proxy) if isinstance(proxy, (str, URL)) else proxy
return {"all://": proxy}
@property
def timeout(self) -> Timeout:
return self._timeout
@ -549,8 +502,8 @@ class BaseClient:
"""
headers = Headers(request.headers)
if not _same_origin(url, request.url):
if not _is_https_redirect(request.url, url):
if not same_origin(url, request.url):
if not is_https_redirect(request.url, url):
# Strip Authorization headers when responses are redirected
# away from the origin. (Except for direct HTTP to HTTPS redirects.)
headers.pop("Authorization", None)
@ -614,12 +567,13 @@ class Client(BaseClient):
sending requests.
* **cookies** - *(optional)* Dictionary of Cookie items to include when
sending requests.
* **verify** - *(optional)* Either `True` to use an SSL context with the
default CA bundle, `False` to disable verification, or an instance of
`ssl.SSLContext` to use a custom context.
* **ssl_context** - *(optional)* An SSL certificate used by the requested host
to authenticate the client.
* **http2** - *(optional)* A boolean indicating if HTTP/2 support should be
enabled. Defaults to `False`.
* **proxy** - *(optional)* A proxy URL where all the traffic should be routed.
* **proxies** - *(optional)* A dictionary mapping proxy keys to proxy
URLs.
* **timeout** - *(optional)* The timeout configuration to use when sending
requests.
* **limits** - *(optional)* The limits configuration to use.
@ -643,13 +597,10 @@ class Client(BaseClient):
params: QueryParamTypes | None = None,
headers: HeaderTypes | None = None,
cookies: CookieTypes | None = None,
verify: ssl.SSLContext | str | bool = True,
cert: CertTypes | None = None,
trust_env: bool = True,
ssl_context: ssl.SSLContext | None = None,
http1: bool = True,
http2: bool = False,
proxy: ProxyTypes | None = None,
mounts: None | (typing.Mapping[str, BaseTransport | None]) = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
follow_redirects: bool = False,
limits: Limits = DEFAULT_LIMITS,
@ -657,7 +608,11 @@ class Client(BaseClient):
event_hooks: None | (typing.Mapping[str, list[EventHook]]) = None,
base_url: URL | str = "",
transport: BaseTransport | None = None,
trust_env: bool = True,
default_encoding: str | typing.Callable[[bytes], str] = "utf-8",
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> None:
super().__init__(
auth=auth,
@ -673,99 +628,21 @@ class Client(BaseClient):
default_encoding=default_encoding,
)
if http2:
try:
import h2 # noqa
except ImportError: # pragma: no cover
raise ImportError(
"Using http2=True, but the 'h2' package is not installed. "
"Make sure to install httpx using `pip install httpx[http2]`."
) from None
allow_env_proxies = trust_env and transport is None
proxy_map = self._get_proxy_map(proxy, allow_env_proxies)
self._transport = self._init_transport(
verify=verify,
cert=cert,
trust_env=trust_env,
http1=http1,
http2=http2,
limits=limits,
transport=transport,
)
self._mounts: dict[URLPattern, BaseTransport | None] = {
URLPattern(key): None
if proxy is None
else self._init_proxy_transport(
proxy,
verify=verify,
cert=cert,
trust_env=trust_env,
if transport is not None:
self._transport = transport
else:
self._transport = HTTPTransport(
ssl_context=ssl_context,
proxy=proxy,
http1=http1,
http2=http2,
limits=limits,
)
for key, proxy in proxy_map.items()
}
if mounts is not None:
self._mounts.update(
{URLPattern(key): transport for key, transport in mounts.items()}
verify=verify,
cert=cert,
)
self._mounts = dict(sorted(self._mounts.items()))
def _init_transport(
self,
verify: ssl.SSLContext | str | bool = True,
cert: CertTypes | None = None,
trust_env: bool = True,
http1: bool = True,
http2: bool = False,
limits: Limits = DEFAULT_LIMITS,
transport: BaseTransport | None = None,
) -> BaseTransport:
if transport is not None:
return transport
return HTTPTransport(
verify=verify,
cert=cert,
trust_env=trust_env,
http1=http1,
http2=http2,
limits=limits,
)
def _init_proxy_transport(
self,
proxy: Proxy,
verify: ssl.SSLContext | str | bool = True,
cert: CertTypes | None = None,
trust_env: bool = True,
http1: bool = True,
http2: bool = False,
limits: Limits = DEFAULT_LIMITS,
) -> BaseTransport:
return HTTPTransport(
verify=verify,
cert=cert,
trust_env=trust_env,
http1=http1,
http2=http2,
limits=limits,
proxy=proxy,
)
def _transport_for_url(self, url: URL) -> BaseTransport:
"""
Returns the transport instance that should be used for a given URL.
This will either be the standard connection pool, or a proxy.
"""
for pattern, transport in self._mounts.items():
if pattern.matches(url):
return self._transport if transport is None else transport
@property
def transport(self) -> BaseTransport:
return self._transport
def request(
@ -1002,7 +879,6 @@ class Client(BaseClient):
"""
Sends a single request, without handling any redirections.
"""
transport = self._transport_for_url(request.url)
start = time.perf_counter()
if not isinstance(request.stream, SyncByteStream):
@ -1011,7 +887,7 @@ class Client(BaseClient):
)
with request_context(request=request):
response = transport.handle_request(request)
response = self.transport.handle_request(request)
assert isinstance(response.stream, SyncByteStream)
@ -1262,15 +1138,11 @@ class Client(BaseClient):
def close(self) -> None:
"""
Close transport and proxies.
Close transport.
"""
if self._state != ClientState.CLOSED:
self._state = ClientState.CLOSED
self._transport.close()
for transport in self._mounts.values():
if transport is not None:
transport.close()
self.transport.close()
def __enter__(self: T) -> T:
if self._state != ClientState.UNOPENED:
@ -1283,11 +1155,7 @@ class Client(BaseClient):
raise RuntimeError(msg)
self._state = ClientState.OPENED
self._transport.__enter__()
for transport in self._mounts.values():
if transport is not None:
transport.__enter__()
self.transport.__enter__()
return self
def __exit__(
@ -1297,11 +1165,7 @@ class Client(BaseClient):
traceback: TracebackType | None = None,
) -> None:
self._state = ClientState.CLOSED
self._transport.__exit__(exc_type, exc_value, traceback)
for transport in self._mounts.values():
if transport is not None:
transport.__exit__(exc_type, exc_value, traceback)
self.transport.__exit__(exc_type, exc_value, traceback)
class AsyncClient(BaseClient):
@ -1328,9 +1192,8 @@ class AsyncClient(BaseClient):
sending requests.
* **cookies** - *(optional)* Dictionary of Cookie items to include when
sending requests.
* **verify** - *(optional)* Either `True` to use an SSL context with the
default CA bundle, `False` to disable verification, or an instance of
`ssl.SSLContext` to use a custom context.
* **ssl_context** - *(optional)* An SSL certificate used by the requested host
to authenticate the client.
* **http2** - *(optional)* A boolean indicating if HTTP/2 support should be
enabled. Defaults to `False`.
* **proxy** - *(optional)* A proxy URL where all the traffic should be routed.
@ -1357,12 +1220,10 @@ class AsyncClient(BaseClient):
params: QueryParamTypes | None = None,
headers: HeaderTypes | None = None,
cookies: CookieTypes | None = None,
verify: ssl.SSLContext | str | bool = True,
cert: CertTypes | None = None,
ssl_context: ssl.SSLContext | None = None,
http1: bool = True,
http2: bool = False,
proxy: ProxyTypes | None = None,
mounts: None | (typing.Mapping[str, AsyncBaseTransport | None]) = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
follow_redirects: bool = False,
limits: Limits = DEFAULT_LIMITS,
@ -1372,6 +1233,9 @@ class AsyncClient(BaseClient):
transport: AsyncBaseTransport | None = None,
trust_env: bool = True,
default_encoding: str | typing.Callable[[bytes], str] = "utf-8",
# Deprecated in favor of `ssl_context`...
verify: typing.Any = None,
cert: typing.Any = None,
) -> None:
super().__init__(
auth=auth,
@ -1386,100 +1250,21 @@ class AsyncClient(BaseClient):
trust_env=trust_env,
default_encoding=default_encoding,
)
if http2:
try:
import h2 # noqa
except ImportError: # pragma: no cover
raise ImportError(
"Using http2=True, but the 'h2' package is not installed. "
"Make sure to install httpx using `pip install httpx[http2]`."
) from None
allow_env_proxies = trust_env and transport is None
proxy_map = self._get_proxy_map(proxy, allow_env_proxies)
self._transport = self._init_transport(
verify=verify,
cert=cert,
trust_env=trust_env,
http1=http1,
http2=http2,
limits=limits,
transport=transport,
)
self._mounts: dict[URLPattern, AsyncBaseTransport | None] = {
URLPattern(key): None
if proxy is None
else self._init_proxy_transport(
proxy,
verify=verify,
cert=cert,
trust_env=trust_env,
if transport is not None:
self._transport = transport
else:
self._transport = AsyncHTTPTransport(
ssl_context=ssl_context,
proxy=proxy,
http1=http1,
http2=http2,
limits=limits,
verify=verify,
cert=cert,
)
for key, proxy in proxy_map.items()
}
if mounts is not None:
self._mounts.update(
{URLPattern(key): transport for key, transport in mounts.items()}
)
self._mounts = dict(sorted(self._mounts.items()))
def _init_transport(
self,
verify: ssl.SSLContext | str | bool = True,
cert: CertTypes | None = None,
trust_env: bool = True,
http1: bool = True,
http2: bool = False,
limits: Limits = DEFAULT_LIMITS,
transport: AsyncBaseTransport | None = None,
) -> AsyncBaseTransport:
if transport is not None:
return transport
return AsyncHTTPTransport(
verify=verify,
cert=cert,
trust_env=trust_env,
http1=http1,
http2=http2,
limits=limits,
)
def _init_proxy_transport(
self,
proxy: Proxy,
verify: ssl.SSLContext | str | bool = True,
cert: CertTypes | None = None,
trust_env: bool = True,
http1: bool = True,
http2: bool = False,
limits: Limits = DEFAULT_LIMITS,
) -> AsyncBaseTransport:
return AsyncHTTPTransport(
verify=verify,
cert=cert,
trust_env=trust_env,
http1=http1,
http2=http2,
limits=limits,
proxy=proxy,
)
def _transport_for_url(self, url: URL) -> AsyncBaseTransport:
"""
Returns the transport instance that should be used for a given URL.
This will either be the standard connection pool, or a proxy.
"""
for pattern, transport in self._mounts.items():
if pattern.matches(url):
return self._transport if transport is None else transport
@property
def transport(self) -> AsyncBaseTransport:
return self._transport
async def request(
@ -1718,16 +1503,15 @@ class AsyncClient(BaseClient):
"""
Sends a single request, without handling any redirections.
"""
transport = self._transport_for_url(request.url)
start = time.perf_counter()
if not isinstance(request.stream, AsyncByteStream):
raise RuntimeError(
"Attempted to send a sync request with an AsyncClient instance."
"Attempted to send an sync request with an AsyncClient instance."
)
with request_context(request=request):
response = await transport.handle_async_request(request)
response = await self.transport.handle_async_request(request)
assert isinstance(response.stream, AsyncByteStream)
response.request = request
@ -1977,15 +1761,11 @@ class AsyncClient(BaseClient):
async def aclose(self) -> None:
"""
Close transport and proxies.
Close transport.
"""
if self._state != ClientState.CLOSED:
self._state = ClientState.CLOSED
await self._transport.aclose()
for proxy in self._mounts.values():
if proxy is not None:
await proxy.aclose()
await self.transport.aclose()
async def __aenter__(self: U) -> U:
if self._state != ClientState.UNOPENED:
@ -1998,11 +1778,7 @@ class AsyncClient(BaseClient):
raise RuntimeError(msg)
self._state = ClientState.OPENED
await self._transport.__aenter__()
for proxy in self._mounts.values():
if proxy is not None:
await proxy.__aenter__()
await self.transport.__aenter__()
return self
async def __aexit__(
@ -2012,8 +1788,4 @@ class AsyncClient(BaseClient):
traceback: TracebackType | None = None,
) -> None:
self._state = ClientState.CLOSED
await self._transport.__aexit__(exc_type, exc_value, traceback)
for proxy in self._mounts.values():
if proxy is not None:
await proxy.__aexit__(exc_type, exc_value, traceback)
await self.transport.__aexit__(exc_type, exc_value, traceback)

View File

@ -1,16 +1,16 @@
from __future__ import annotations
import os
import ssl
import sys
import typing
import warnings
from ._models import Headers
from ._types import CertTypes, HeaderTypes, TimeoutTypes
from ._types import HeaderTypes, TimeoutTypes
from ._urls import URL
if typing.TYPE_CHECKING:
import ssl # pragma: no cover
__all__ = ["Limits", "Proxy", "Timeout", "create_ssl_context"]
__all__ = ["Limits", "Proxy", "SSLContext", "Timeout", "create_ssl_context"]
class UnsetType:
@ -21,52 +21,104 @@ UNSET = UnsetType()
def create_ssl_context(
verify: ssl.SSLContext | str | bool = True,
cert: CertTypes | None = None,
verify: typing.Any = None,
cert: typing.Any = None,
trust_env: bool = True,
) -> ssl.SSLContext:
import ssl
import warnings
import certifi
if verify is True:
if trust_env and os.environ.get("SSL_CERT_FILE"): # pragma: nocover
ctx = ssl.create_default_context(cafile=os.environ["SSL_CERT_FILE"])
elif trust_env and os.environ.get("SSL_CERT_DIR"): # pragma: nocover
ctx = ssl.create_default_context(capath=os.environ["SSL_CERT_DIR"])
else:
# Default case...
ctx = ssl.create_default_context(cafile=certifi.where())
elif verify is False:
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
elif isinstance(verify, str): # pragma: nocover
message = (
"`verify=<str>` is deprecated. "
"Use `verify=ssl.create_default_context(cafile=...)` "
"or `verify=ssl.create_default_context(capath=...)` instead."
http2: bool = False,
) -> ssl.SSLContext: # pragma: nocover
# The `create_ssl_context` helper function is now deprecated
# in favour of `httpx.SSLContext()`.
if isinstance(verify, bool):
ssl_context: ssl.SSLContext = SSLContext(verify=verify)
warnings.warn(
"The verify=<bool> parameter is deprecated since 0.28.0. "
"Use `ssl_context=httpx.SSLContext(verify=<bool>)`."
)
warnings.warn(message, DeprecationWarning)
if os.path.isdir(verify):
return ssl.create_default_context(capath=verify)
return ssl.create_default_context(cafile=verify)
elif isinstance(verify, str):
warnings.warn(
"The verify=<str> parameter is deprecated since 0.28.0. "
"Use `ssl_context=httpx.SSLContext()` and `.load_verify_locations()`."
)
ssl_context = SSLContext()
if os.path.isfile(verify):
ssl_context.load_verify_locations(cafile=verify)
elif os.path.isdir(verify):
ssl_context.load_verify_locations(capath=verify)
elif isinstance(verify, ssl.SSLContext):
warnings.warn(
"The verify=<ssl context> parameter is deprecated since 0.28.0. "
"Use `ssl_context = httpx.SSLContext()`."
)
ssl_context = verify
else:
ctx = verify
if cert: # pragma: nocover
message = (
"`cert=...` is deprecated. Use `verify=<ssl_context>` instead,"
"with `.load_cert_chain()` to configure the certificate chain."
warnings.warn(
"`create_ssl_context()` is deprecated since 0.28.0."
"Use `ssl_context = httpx.SSLContext()`."
)
warnings.warn(message, DeprecationWarning)
if isinstance(cert, str):
ctx.load_cert_chain(cert)
else:
ctx.load_cert_chain(*cert)
ssl_context = SSLContext()
return ctx
if cert is not None:
warnings.warn(
"The `cert=<...>` parameter is deprecated since 0.28.0. "
"Use `ssl_context = httpx.SSLContext()` and `.load_cert_chain()`."
)
if isinstance(cert, str):
ssl_context.load_cert_chain(cert)
else:
ssl_context.load_cert_chain(*cert)
return ssl_context
class SSLContext(ssl.SSLContext):
def __init__(
self,
verify: bool = True,
) -> None:
import certifi
# ssl.SSLContext sets OP_NO_SSLv2, OP_NO_SSLv3, OP_NO_COMPRESSION,
# OP_CIPHER_SERVER_PREFERENCE, OP_SINGLE_DH_USE and OP_SINGLE_ECDH_USE
# by default. (from `ssl.create_default_context`)
super().__init__()
self._verify = verify
# Our SSL setup here is similar to the stdlib `ssl.create_default_context()`
# implementation, except with `certifi` used for certificate verification.
if not verify:
self.check_hostname = False
self.verify_mode = ssl.CERT_NONE
return
self.verify_mode = ssl.CERT_REQUIRED
self.check_hostname = True
# Use stricter verify flags where possible.
if hasattr(ssl, "VERIFY_X509_PARTIAL_CHAIN"): # pragma: nocover
self.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN
if hasattr(ssl, "VERIFY_X509_STRICT"): # pragma: nocover
self.verify_flags |= ssl.VERIFY_X509_STRICT
# Default to `certifi` for certificiate verification.
self.load_verify_locations(cafile=certifi.where())
# OpenSSL keylog file support.
if hasattr(self, "keylog_filename"):
keylogfile = os.environ.get("SSLKEYLOGFILE")
if keylogfile and not sys.flags.ignore_environment:
self.keylog_filename = keylogfile
def __repr__(self) -> str:
class_name = self.__class__.__name__
return f"<{class_name}(verify={self._verify!r})>"
def __new__(
cls,
protocol: ssl._SSLMethod = ssl.PROTOCOL_TLS_CLIENT,
*args: typing.Any,
**kwargs: typing.Any,
) -> "SSLContext":
return super().__new__(cls, protocol, *args, **kwargs)
class Timeout:
@ -223,15 +275,6 @@ class Proxy:
self.headers = headers
self.ssl_context = ssl_context
@property
def raw_auth(self) -> tuple[bytes, bytes] | None:
# The proxy authentication as raw bytes.
return (
None
if self.auth is None
else (self.auth[0].encode("utf-8"), self.auth[1].encode("utf-8"))
)
def __repr__(self) -> str:
# The authentication is represented with the password component masked.
auth = (self.auth[0], "********") if self.auth else None

View File

@ -175,11 +175,9 @@ class ZStandardDecoder(ContentDecoder):
) from None
self.decompressor = zstandard.ZstdDecompressor().decompressobj()
self.seen_data = False
def decode(self, data: bytes) -> bytes:
assert zstandard is not None
self.seen_data = True
output = io.BytesIO()
try:
output.write(self.decompressor.decompress(data))
@ -192,8 +190,6 @@ class ZStandardDecoder(ContentDecoder):
return output.getvalue()
def flush(self) -> bytes:
if not self.seen_data:
return b""
ret = self.decompressor.flush() # note: this is a no-op
if not self.decompressor.eof:
raise DecodingError("Zstandard data is incomplete") # pragma: no cover

View File

@ -331,7 +331,9 @@ class StreamClosed(StreamError):
"""
def __init__(self) -> None:
message = "Attempted to read or stream content, but the stream has been closed."
message = (
"Attempted to read or stream content, but the stream has " "been closed."
)
super().__init__(message)

View File

@ -15,6 +15,7 @@ import rich.syntax
import rich.table
from ._client import Client
from ._config import SSLContext
from ._exceptions import RequestError
from ._models import Response
from ._status_codes import codes
@ -475,8 +476,11 @@ def main(
if not method:
method = "POST" if content or data or files or json else "GET"
ssl_context = SSLContext(verify=verify)
try:
with Client(proxy=proxy, timeout=timeout, http2=http2, verify=verify) as client:
with Client(
proxy=proxy, timeout=timeout, http2=http2, ssl_context=ssl_context
) as client:
with client.stream(
method,
url,

View File

@ -1,10 +1,8 @@
from __future__ import annotations
import codecs
import datetime
import email.message
import json as jsonlib
import re
import typing
import urllib.request
from collections.abc import Mapping
@ -46,95 +44,17 @@ from ._types import (
SyncByteStream,
)
from ._urls import URL
from ._utils import to_bytes_or_str, to_str
from ._utils import (
is_known_encoding,
normalize_header_key,
normalize_header_value,
obfuscate_sensitive_headers,
parse_content_type_charset,
parse_header_links,
)
__all__ = ["Cookies", "Headers", "Request", "Response"]
SENSITIVE_HEADERS = {"authorization", "proxy-authorization"}
def _is_known_encoding(encoding: str) -> bool:
"""
Return `True` if `encoding` is a known codec.
"""
try:
codecs.lookup(encoding)
except LookupError:
return False
return True
def _normalize_header_key(key: str | bytes, encoding: str | None = None) -> bytes:
"""
Coerce str/bytes into a strictly byte-wise HTTP header key.
"""
return key if isinstance(key, bytes) else key.encode(encoding or "ascii")
def _normalize_header_value(value: str | bytes, encoding: str | None = None) -> bytes:
"""
Coerce str/bytes into a strictly byte-wise HTTP header value.
"""
if isinstance(value, bytes):
return value
if not isinstance(value, str):
raise TypeError(f"Header value must be str or bytes, not {type(value)}")
return value.encode(encoding or "ascii")
def _parse_content_type_charset(content_type: str) -> str | None:
# We used to use `cgi.parse_header()` here, but `cgi` became a dead battery.
# See: https://peps.python.org/pep-0594/#cgi
msg = email.message.Message()
msg["content-type"] = content_type
return msg.get_content_charset(failobj=None)
def _parse_header_links(value: str) -> list[dict[str, str]]:
"""
Returns a list of parsed link headers, for more info see:
https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Link
The generic syntax of those is:
Link: < uri-reference >; param1=value1; param2="value2"
So for instance:
Link; '<http:/.../front.jpeg>; type="image/jpeg",<http://.../back.jpeg>;'
would return
[
{"url": "http:/.../front.jpeg", "type": "image/jpeg"},
{"url": "http://.../back.jpeg"},
]
:param value: HTTP Link entity-header field
:return: list of parsed link headers
"""
links: list[dict[str, str]] = []
replace_chars = " '\""
value = value.strip(replace_chars)
if not value:
return links
for val in re.split(", *<", value):
try:
url, params = val.split(";", 1)
except ValueError:
url, params = val, ""
link = {"url": url.strip("<> '\"")}
for param in params.split(";"):
try:
key, value = param.split("=")
except ValueError:
break
link[key.strip(replace_chars)] = value.strip(replace_chars)
links.append(link)
return links
def _obfuscate_sensitive_headers(
items: typing.Iterable[tuple[typing.AnyStr, typing.AnyStr]],
) -> typing.Iterator[tuple[typing.AnyStr, typing.AnyStr]]:
for k, v in items:
if to_str(k.lower()) in SENSITIVE_HEADERS:
v = to_bytes_or_str("[secure]", match_type_of=v)
yield k, v
class Headers(typing.MutableMapping[str, str]):
"""
@ -146,20 +66,28 @@ class Headers(typing.MutableMapping[str, str]):
headers: HeaderTypes | None = None,
encoding: str | None = None,
) -> None:
self._list = [] # type: typing.List[typing.Tuple[bytes, bytes, bytes]]
if isinstance(headers, Headers):
if headers is None:
self._list = [] # type: typing.List[typing.Tuple[bytes, bytes, bytes]]
elif isinstance(headers, Headers):
self._list = list(headers._list)
elif isinstance(headers, Mapping):
for k, v in headers.items():
bytes_key = _normalize_header_key(k, encoding)
bytes_value = _normalize_header_value(v, encoding)
self._list.append((bytes_key, bytes_key.lower(), bytes_value))
elif headers is not None:
for k, v in headers:
bytes_key = _normalize_header_key(k, encoding)
bytes_value = _normalize_header_value(v, encoding)
self._list.append((bytes_key, bytes_key.lower(), bytes_value))
self._list = [
(
normalize_header_key(k, lower=False, encoding=encoding),
normalize_header_key(k, lower=True, encoding=encoding),
normalize_header_value(v, encoding),
)
for k, v in headers.items()
]
else:
self._list = [
(
normalize_header_key(k, lower=False, encoding=encoding),
normalize_header_key(k, lower=True, encoding=encoding),
normalize_header_value(v, encoding),
)
for k, v in headers
]
self._encoding = encoding
@ -370,7 +298,7 @@ class Headers(typing.MutableMapping[str, str]):
if self.encoding != "ascii":
encoding_str = f", encoding={self.encoding!r}"
as_list = list(_obfuscate_sensitive_headers(self.multi_items()))
as_list = list(obfuscate_sensitive_headers(self.multi_items()))
as_dict = dict(as_list)
no_duplicate_keys = len(as_dict) == len(as_list)
@ -398,7 +326,7 @@ class Request:
self.method = method.upper()
self.url = URL(url) if params is None else URL(url, params=params)
self.headers = Headers(headers)
self.extensions = {} if extensions is None else dict(extensions)
self.extensions = {} if extensions is None else extensions
if cookies:
Cookies(cookies).set_cookie_header(self)
@ -537,7 +465,7 @@ class Response:
# the client will set `response.next_request`.
self.next_request: Request | None = None
self.extensions = {} if extensions is None else dict(extensions)
self.extensions: ResponseExtensions = {} if extensions is None else extensions
self.history = [] if history is None else list(history)
self.is_closed = False
@ -663,7 +591,7 @@ class Response:
"""
if not hasattr(self, "_encoding"):
encoding = self.charset_encoding
if encoding is None or not _is_known_encoding(encoding):
if encoding is None or not is_known_encoding(encoding):
if isinstance(self.default_encoding, str):
encoding = self.default_encoding
elif hasattr(self, "_content"):
@ -694,7 +622,7 @@ class Response:
if content_type is None:
return None
return _parse_content_type_charset(content_type)
return parse_content_type_charset(content_type)
def _get_content_decoder(self) -> ContentDecoder:
"""
@ -849,7 +777,7 @@ class Response:
return {
(link.get("rel") or link.get("url")): link
for link in _parse_header_links(header)
for link in parse_header_links(header)
}
@property
@ -964,7 +892,7 @@ class Response:
Automatically called if the response body is read to completion.
"""
if not isinstance(self.stream, SyncByteStream):
raise RuntimeError("Attempted to call a sync close on an async stream.")
raise RuntimeError("Attempted to call an sync close on an async stream.")
if not self.is_closed:
self.is_closed = True
@ -1045,7 +973,7 @@ class Response:
if self.is_closed:
raise StreamClosed()
if not isinstance(self.stream, AsyncByteStream):
raise RuntimeError("Attempted to call an async iterator on a sync stream.")
raise RuntimeError("Attempted to call an async iterator on an sync stream.")
self.is_stream_consumed = True
self._num_bytes_downloaded = 0
@ -1068,7 +996,7 @@ class Response:
Automatically called if the response body is read to completion.
"""
if not isinstance(self.stream, AsyncByteStream):
raise RuntimeError("Attempted to call an async close on a sync stream.")
raise RuntimeError("Attempted to call an async close on an sync stream.")
if not self.is_closed:
self.is_closed = True

View File

@ -1,9 +1,7 @@
from __future__ import annotations
import io
import mimetypes
import os
import re
import typing
from pathlib import Path
@ -16,42 +14,13 @@ from ._types import (
SyncByteStream,
)
from ._utils import (
format_form_param,
guess_content_type,
peek_filelike_length,
primitive_value_to_str,
to_bytes,
)
_HTML5_FORM_ENCODING_REPLACEMENTS = {'"': "%22", "\\": "\\\\"}
_HTML5_FORM_ENCODING_REPLACEMENTS.update(
{chr(c): "%{:02X}".format(c) for c in range(0x1F + 1) if c != 0x1B}
)
_HTML5_FORM_ENCODING_RE = re.compile(
r"|".join([re.escape(c) for c in _HTML5_FORM_ENCODING_REPLACEMENTS.keys()])
)
def _format_form_param(name: str, value: str) -> bytes:
"""
Encode a name/value pair within a multipart form.
"""
def replacer(match: typing.Match[str]) -> str:
return _HTML5_FORM_ENCODING_REPLACEMENTS[match.group(0)]
value = _HTML5_FORM_ENCODING_RE.sub(replacer, value)
return f'{name}="{value}"'.encode()
def _guess_content_type(filename: str | None) -> str | None:
"""
Guesses the mimetype based on a filename. Defaults to `application/octet-stream`.
Returns `None` if `filename` is `None` or empty.
"""
if filename:
return mimetypes.guess_type(filename)[0] or "application/octet-stream"
return None
def get_multipart_boundary_from_content_type(
content_type: bytes | None,
@ -89,7 +58,7 @@ class DataField:
def render_headers(self) -> bytes:
if not hasattr(self, "_headers"):
name = _format_form_param("name", self.name)
name = format_form_param("name", self.name)
self._headers = b"".join(
[b"Content-Disposition: form-data; ", name, b"\r\n\r\n"]
)
@ -146,7 +115,7 @@ class FileField:
fileobj = value
if content_type is None:
content_type = _guess_content_type(filename)
content_type = guess_content_type(filename)
has_content_type_header = any("content-type" in key.lower() for key in headers)
if content_type is not None and not has_content_type_header:
@ -187,10 +156,10 @@ class FileField:
if not hasattr(self, "_headers"):
parts = [
b"Content-Disposition: form-data; ",
_format_form_param("name", self.name),
format_form_param("name", self.name),
]
if self.filename:
filename = _format_form_param("filename", self.filename)
filename = format_form_param("filename", self.filename)
parts.extend([b"; ", filename])
for header_name, header_value in self.headers.items():
key, val = f"\r\n{header_name}: ".encode(), header_value.encode()

View File

@ -35,7 +35,7 @@ if typing.TYPE_CHECKING:
import httpx # pragma: no cover
from .._config import DEFAULT_LIMITS, Limits, Proxy, create_ssl_context
from .._config import DEFAULT_LIMITS, Limits, Proxy, SSLContext, create_ssl_context
from .._exceptions import (
ConnectError,
ConnectTimeout,
@ -53,7 +53,7 @@ from .._exceptions import (
WriteTimeout,
)
from .._models import Request, Response
from .._types import AsyncByteStream, CertTypes, ProxyTypes, SyncByteStream
from .._types import AsyncByteStream, ProxyTypes, SyncByteStream
from .._urls import URL
from .base import AsyncBaseTransport, BaseTransport
@ -135,9 +135,7 @@ class ResponseStream(SyncByteStream):
class HTTPTransport(BaseTransport):
def __init__(
self,
verify: ssl.SSLContext | str | bool = True,
cert: CertTypes | None = None,
trust_env: bool = True,
ssl_context: ssl.SSLContext | None = None,
http1: bool = True,
http2: bool = False,
limits: Limits = DEFAULT_LIMITS,
@ -146,11 +144,27 @@ class HTTPTransport(BaseTransport):
local_address: str | None = None,
retries: int = 0,
socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
# Deprecated...
verify: typing.Any = None,
cert: typing.Any = None,
) -> None:
if http2:
try:
import h2 # noqa
except ImportError: # pragma: no cover
raise ImportError(
"Using http2=True, but the 'h2' package is not installed. "
"Make sure to install httpx using `pip install httpx[http2]`."
) from None
import httpcore
proxy = Proxy(url=proxy) if isinstance(proxy, (str, URL)) else proxy
ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env)
if verify is not None or cert is not None: # pragma: nocover
# Deprecated...
ssl_context = create_ssl_context(verify, cert)
else:
ssl_context = ssl_context or SSLContext()
if proxy is None:
self._pool = httpcore.ConnectionPool(
@ -279,9 +293,7 @@ class AsyncResponseStream(AsyncByteStream):
class AsyncHTTPTransport(AsyncBaseTransport):
def __init__(
self,
verify: ssl.SSLContext | str | bool = True,
cert: CertTypes | None = None,
trust_env: bool = True,
ssl_context: ssl.SSLContext | None = None,
http1: bool = True,
http2: bool = False,
limits: Limits = DEFAULT_LIMITS,
@ -290,11 +302,27 @@ class AsyncHTTPTransport(AsyncBaseTransport):
local_address: str | None = None,
retries: int = 0,
socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
# Deprecated...
verify: typing.Any = None,
cert: typing.Any = None,
) -> None:
if http2:
try:
import h2 # noqa
except ImportError: # pragma: no cover
raise ImportError(
"Using http2=True, but the 'h2' package is not installed. "
"Make sure to install httpx using `pip install httpx[http2]`."
) from None
import httpcore
proxy = Proxy(url=proxy) if isinstance(proxy, (str, URL)) else proxy
ssl_context = create_ssl_context(verify=verify, cert=cert, trust_env=trust_env)
if verify is not None or cert is not None: # pragma: nocover
# Deprecated...
ssl_context = create_ssl_context(verify, cert)
else:
ssl_context = ssl_context or SSLContext()
if proxy is None:
self._pool = httpcore.AsyncConnectionPool(
@ -355,7 +383,7 @@ class AsyncHTTPTransport(AsyncBaseTransport):
else: # pragma: no cover
raise ValueError(
"Proxy protocol must be either 'http', 'https', 'socks5', or 'socks5h',"
f" but got {proxy.url.scheme!r}."
" but got {proxy.url.scheme!r}."
)
async def __aenter__(self: A) -> A: # Use generics for subclass support.

View File

@ -15,6 +15,7 @@ from typing import (
Iterator,
List,
Mapping,
MutableMapping,
Optional,
Sequence,
Tuple,
@ -57,7 +58,6 @@ TimeoutTypes = Union[
"Timeout",
]
ProxyTypes = Union["URL", str, "Proxy"]
CertTypes = Union[str, Tuple[str, str], Tuple[str, str, str]]
AuthTypes = Union[
Tuple[Union[str, bytes], Union[str, bytes]],
@ -67,7 +67,7 @@ AuthTypes = Union[
RequestContent = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]
ResponseContent = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]
ResponseExtensions = Mapping[str, Any]
ResponseExtensions = MutableMapping[str, Any]
RequestData = Mapping[str, Any]
@ -84,7 +84,7 @@ FileTypes = Union[
]
RequestFiles = Union[Mapping[str, FileTypes], Sequence[Tuple[str, FileTypes]]]
RequestExtensions = Mapping[str, Any]
RequestExtensions = MutableMapping[str, Any]
__all__ = ["AsyncByteStream", "SyncByteStream"]

View File

@ -379,7 +379,7 @@ class URL:
if ":" in userinfo:
# Mask any password component.
userinfo = f"{userinfo.split(':')[0]}:[secure]"
userinfo = f'{userinfo.split(":")[0]}:[secure]'
authority = "".join(
[
@ -400,22 +400,6 @@ class URL:
return f"{self.__class__.__name__}({url!r})"
@property
def raw(self) -> tuple[bytes, bytes, int, bytes]: # pragma: nocover
import collections
import warnings
warnings.warn("URL.raw is deprecated.")
RawURL = collections.namedtuple(
"RawURL", ["raw_scheme", "raw_host", "port", "raw_path"]
)
return RawURL(
raw_scheme=self.raw_scheme,
raw_host=self.raw_host,
port=self.port,
raw_path=self.raw_path,
)
class QueryParams(typing.Mapping[str, str]):
"""

View File

@ -1,10 +1,11 @@
from __future__ import annotations
import ipaddress
import codecs
import email.message
import mimetypes
import os
import re
import typing
from urllib.request import getproxies
from ._types import PrimitiveData
@ -12,6 +13,42 @@ if typing.TYPE_CHECKING: # pragma: no cover
from ._urls import URL
_HTML5_FORM_ENCODING_REPLACEMENTS = {'"': "%22", "\\": "\\\\"}
_HTML5_FORM_ENCODING_REPLACEMENTS.update(
{chr(c): "%{:02X}".format(c) for c in range(0x1F + 1) if c != 0x1B}
)
_HTML5_FORM_ENCODING_RE = re.compile(
r"|".join([re.escape(c) for c in _HTML5_FORM_ENCODING_REPLACEMENTS.keys()])
)
def normalize_header_key(
value: str | bytes,
lower: bool,
encoding: str | None = None,
) -> bytes:
"""
Coerce str/bytes into a strictly byte-wise HTTP header key.
"""
if isinstance(value, bytes):
bytes_value = value
else:
bytes_value = value.encode(encoding or "ascii")
return bytes_value.lower() if lower else bytes_value
def normalize_header_value(value: str | bytes, encoding: str | None = None) -> bytes:
"""
Coerce str/bytes into a strictly byte-wise HTTP header value.
"""
if isinstance(value, bytes):
return value
if not isinstance(value, str):
raise TypeError(f"Header value must be str or bytes, not {type(value)}")
return value.encode(encoding or "ascii")
def primitive_value_to_str(value: PrimitiveData) -> str:
"""
Coerce a primitive data type into a string value.
@ -27,53 +64,116 @@ def primitive_value_to_str(value: PrimitiveData) -> str:
return str(value)
def get_environment_proxies() -> dict[str, str | None]:
"""Gets proxy information from the environment"""
def is_known_encoding(encoding: str) -> bool:
"""
Return `True` if `encoding` is a known codec.
"""
try:
codecs.lookup(encoding)
except LookupError:
return False
return True
# urllib.request.getproxies() falls back on System
# Registry and Config for proxies on Windows and macOS.
# We don't want to propagate non-HTTP proxies into
# our configuration such as 'TRAVIS_APT_PROXY'.
proxy_info = getproxies()
mounts: dict[str, str | None] = {}
for scheme in ("http", "https", "all"):
if proxy_info.get(scheme):
hostname = proxy_info[scheme]
mounts[f"{scheme}://"] = (
hostname if "://" in hostname else f"http://{hostname}"
)
def format_form_param(name: str, value: str) -> bytes:
"""
Encode a name/value pair within a multipart form.
"""
no_proxy_hosts = [host.strip() for host in proxy_info.get("no", "").split(",")]
for hostname in no_proxy_hosts:
# See https://curl.haxx.se/libcurl/c/CURLOPT_NOPROXY.html for details
# on how names in `NO_PROXY` are handled.
if hostname == "*":
# If NO_PROXY=* is used or if "*" occurs as any one of the comma
# separated hostnames, then we should just bypass any information
# from HTTP_PROXY, HTTPS_PROXY, ALL_PROXY, and always ignore
# proxies.
return {}
elif hostname:
# NO_PROXY=.google.com is marked as "all://*.google.com,
# which disables "www.google.com" but not "google.com"
# NO_PROXY=google.com is marked as "all://*google.com,
# which disables "www.google.com" and "google.com".
# (But not "wwwgoogle.com")
# NO_PROXY can include domains, IPv6, IPv4 addresses and "localhost"
# NO_PROXY=example.com,::1,localhost,192.168.0.0/16
if "://" in hostname:
mounts[hostname] = None
elif is_ipv4_hostname(hostname):
mounts[f"all://{hostname}"] = None
elif is_ipv6_hostname(hostname):
mounts[f"all://[{hostname}]"] = None
elif hostname.lower() == "localhost":
mounts[f"all://{hostname}"] = None
else:
mounts[f"all://*{hostname}"] = None
def replacer(match: typing.Match[str]) -> str:
return _HTML5_FORM_ENCODING_REPLACEMENTS[match.group(0)]
return mounts
value = _HTML5_FORM_ENCODING_RE.sub(replacer, value)
return f'{name}="{value}"'.encode()
def parse_header_links(value: str) -> list[dict[str, str]]:
"""
Returns a list of parsed link headers, for more info see:
https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Link
The generic syntax of those is:
Link: < uri-reference >; param1=value1; param2="value2"
So for instance:
Link; '<http:/.../front.jpeg>; type="image/jpeg",<http://.../back.jpeg>;'
would return
[
{"url": "http:/.../front.jpeg", "type": "image/jpeg"},
{"url": "http://.../back.jpeg"},
]
:param value: HTTP Link entity-header field
:return: list of parsed link headers
"""
links: list[dict[str, str]] = []
replace_chars = " '\""
value = value.strip(replace_chars)
if not value:
return links
for val in re.split(", *<", value):
try:
url, params = val.split(";", 1)
except ValueError:
url, params = val, ""
link = {"url": url.strip("<> '\"")}
for param in params.split(";"):
try:
key, value = param.split("=")
except ValueError:
break
link[key.strip(replace_chars)] = value.strip(replace_chars)
links.append(link)
return links
def parse_content_type_charset(content_type: str) -> str | None:
# We used to use `cgi.parse_header()` here, but `cgi` became a dead battery.
# See: https://peps.python.org/pep-0594/#cgi
msg = email.message.Message()
msg["content-type"] = content_type
return msg.get_content_charset(failobj=None)
SENSITIVE_HEADERS = {"authorization", "proxy-authorization"}
def obfuscate_sensitive_headers(
items: typing.Iterable[tuple[typing.AnyStr, typing.AnyStr]],
) -> typing.Iterator[tuple[typing.AnyStr, typing.AnyStr]]:
for k, v in items:
if to_str(k.lower()) in SENSITIVE_HEADERS:
v = to_bytes_or_str("[secure]", match_type_of=v)
yield k, v
def port_or_default(url: URL) -> int | None:
if url.port is not None:
return url.port
return {"http": 80, "https": 443}.get(url.scheme)
def same_origin(url: URL, other: URL) -> bool:
"""
Return 'True' if the given URLs share the same origin.
"""
return (
url.scheme == other.scheme
and url.host == other.host
and port_or_default(url) == port_or_default(other)
)
def is_https_redirect(url: URL, location: URL) -> bool:
"""
Return 'True' if 'location' is a HTTPS upgrade of 'url'
"""
if url.host != location.host:
return False
return (
url.scheme == "http"
and port_or_default(url) == 80
and location.scheme == "https"
and port_or_default(location) == 443
)
def to_bytes(value: str | bytes, encoding: str = "utf-8") -> bytes:
@ -92,6 +192,12 @@ def unquote(value: str) -> str:
return value[1:-1] if value[0] == value[-1] == '"' else value
def guess_content_type(filename: str | None) -> str | None:
if filename:
return mimetypes.guess_type(filename)[0] or "application/octet-stream"
return None
def peek_filelike_length(stream: typing.Any) -> int | None:
"""
Given a file-like stream object, return its length in number of bytes
@ -115,128 +221,3 @@ def peek_filelike_length(stream: typing.Any) -> int | None:
return None
return length
class URLPattern:
"""
A utility class currently used for making lookups against proxy keys...
# Wildcard matching...
>>> pattern = URLPattern("all://")
>>> pattern.matches(httpx.URL("http://example.com"))
True
# Witch scheme matching...
>>> pattern = URLPattern("https://")
>>> pattern.matches(httpx.URL("https://example.com"))
True
>>> pattern.matches(httpx.URL("http://example.com"))
False
# With domain matching...
>>> pattern = URLPattern("https://example.com")
>>> pattern.matches(httpx.URL("https://example.com"))
True
>>> pattern.matches(httpx.URL("http://example.com"))
False
>>> pattern.matches(httpx.URL("https://other.com"))
False
# Wildcard scheme, with domain matching...
>>> pattern = URLPattern("all://example.com")
>>> pattern.matches(httpx.URL("https://example.com"))
True
>>> pattern.matches(httpx.URL("http://example.com"))
True
>>> pattern.matches(httpx.URL("https://other.com"))
False
# With port matching...
>>> pattern = URLPattern("https://example.com:1234")
>>> pattern.matches(httpx.URL("https://example.com:1234"))
True
>>> pattern.matches(httpx.URL("https://example.com"))
False
"""
def __init__(self, pattern: str) -> None:
from ._urls import URL
if pattern and ":" not in pattern:
raise ValueError(
f"Proxy keys should use proper URL forms rather "
f"than plain scheme strings. "
f'Instead of "{pattern}", use "{pattern}://"'
)
url = URL(pattern)
self.pattern = pattern
self.scheme = "" if url.scheme == "all" else url.scheme
self.host = "" if url.host == "*" else url.host
self.port = url.port
if not url.host or url.host == "*":
self.host_regex: typing.Pattern[str] | None = None
elif url.host.startswith("*."):
# *.example.com should match "www.example.com", but not "example.com"
domain = re.escape(url.host[2:])
self.host_regex = re.compile(f"^.+\\.{domain}$")
elif url.host.startswith("*"):
# *example.com should match "www.example.com" and "example.com"
domain = re.escape(url.host[1:])
self.host_regex = re.compile(f"^(.+\\.)?{domain}$")
else:
# example.com should match "example.com" but not "www.example.com"
domain = re.escape(url.host)
self.host_regex = re.compile(f"^{domain}$")
def matches(self, other: URL) -> bool:
if self.scheme and self.scheme != other.scheme:
return False
if (
self.host
and self.host_regex is not None
and not self.host_regex.match(other.host)
):
return False
if self.port is not None and self.port != other.port:
return False
return True
@property
def priority(self) -> tuple[int, int, int]:
"""
The priority allows URLPattern instances to be sortable, so that
we can match from most specific to least specific.
"""
# URLs with a port should take priority over URLs without a port.
port_priority = 0 if self.port is not None else 1
# Longer hostnames should match first.
host_priority = -len(self.host)
# Longer schemes should match first.
scheme_priority = -len(self.scheme)
return (port_priority, host_priority, scheme_priority)
def __hash__(self) -> int:
return hash(self.pattern)
def __lt__(self, other: URLPattern) -> bool:
return self.priority < other.priority
def __eq__(self, other: typing.Any) -> bool:
return isinstance(other, URLPattern) and self.pattern == other.pattern
def is_ipv4_hostname(hostname: str) -> bool:
try:
ipaddress.IPv4Address(hostname.split("/")[0])
except Exception:
return False
return True
def is_ipv6_hostname(hostname: str) -> bool:
try:
ipaddress.IPv6Address(hostname.split("/")[0])
except Exception:
return False
return True

View File

@ -6,7 +6,7 @@ build-backend = "hatchling.build"
name = "httpx"
description = "The next generation HTTP client."
license = "BSD-3-Clause"
requires-python = ">=3.9"
requires-python = ">=3.8"
authors = [
{ name = "Tom Christie", email = "tom@tomchristie.com" },
]
@ -20,11 +20,11 @@ classifiers = [
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Topic :: Internet :: WWW/HTTP",
]
dependencies = [
@ -43,7 +43,7 @@ brotli = [
cli = [
"click==8.*",
"pygments==2.*",
"rich>=10,<15",
"rich>=10,<14",
]
http2 = [
"h2>=3,<5",

View File

@ -11,19 +11,19 @@ chardet==5.2.0
# Documentation
mkdocs==1.6.1
mkautodoc==0.2.0
mkdocs-material==9.6.18
mkdocs-material==9.5.39
# Packaging
build==1.3.0
twine==6.1.0
build==1.2.2
twine==5.1.1
# Tests & Linting
coverage[toml]==7.10.6
cryptography==45.0.7
mypy==1.17.1
pytest==8.4.1
ruff==0.12.11
trio==0.31.0
coverage[toml]==7.6.1
cryptography==43.0.1
mypy==1.11.2
pytest==8.3.3
ruff==0.6.8
trio==0.26.2
trio-typing==0.10.0
trustme==1.2.1
uvicorn==0.35.0
trustme==1.1.0
uvicorn==0.31.0

View File

@ -8,5 +8,5 @@ export SOURCE_FILES="httpx tests"
set -x
${PREFIX}ruff check --fix $SOURCE_FILES
${PREFIX}ruff --fix $SOURCE_FILES
${PREFIX}ruff format $SOURCE_FILES

View File

@ -211,47 +211,6 @@ async def test_context_managed_transport():
]
@pytest.mark.anyio
async def test_context_managed_transport_and_mount():
class Transport(httpx.AsyncBaseTransport):
def __init__(self, name: str) -> None:
self.name: str = name
self.events: list[str] = []
async def aclose(self):
# The base implementation of httpx.AsyncBaseTransport just
# calls into `.aclose`, so simple transport cases can just override
# this method for any cleanup, where more complex cases
# might want to additionally override `__aenter__`/`__aexit__`.
self.events.append(f"{self.name}.aclose")
async def __aenter__(self):
await super().__aenter__()
self.events.append(f"{self.name}.__aenter__")
async def __aexit__(self, *args):
await super().__aexit__(*args)
self.events.append(f"{self.name}.__aexit__")
transport = Transport(name="transport")
mounted = Transport(name="mounted")
async with httpx.AsyncClient(
transport=transport, mounts={"http://www.example.org": mounted}
):
pass
assert transport.events == [
"transport.__aenter__",
"transport.aclose",
"transport.__aexit__",
]
assert mounted.events == [
"mounted.__aenter__",
"mounted.aclose",
"mounted.__aexit__",
]
def hello_world(request):
return httpx.Response(200, text="Hello, world!")
@ -288,31 +247,6 @@ async def test_client_closed_state_using_with_block():
await client.get("http://example.com")
def unmounted(request: httpx.Request) -> httpx.Response:
data = {"app": "unmounted"}
return httpx.Response(200, json=data)
def mounted(request: httpx.Request) -> httpx.Response:
data = {"app": "mounted"}
return httpx.Response(200, json=data)
@pytest.mark.anyio
async def test_mounted_transport():
transport = httpx.MockTransport(unmounted)
mounts = {"custom://": httpx.MockTransport(mounted)}
async with httpx.AsyncClient(transport=transport, mounts=mounts) as client:
response = await client.get("https://www.example.com")
assert response.status_code == 200
assert response.json() == {"app": "unmounted"}
response = await client.get("custom://www.example.com")
assert response.status_code == 200
assert response.json() == {"app": "mounted"}
@pytest.mark.anyio
async def test_async_mock_transport():
async def hello_world(request: httpx.Request) -> httpx.Response:

View File

@ -326,7 +326,7 @@ async def test_auth_property() -> None:
async with httpx.AsyncClient(transport=httpx.MockTransport(app)) as client:
assert client.auth is None
client.auth = ("user", "password123")
client.auth = ("user", "password123") # type: ignore
assert isinstance(client.auth, httpx.BasicAuth)
url = "https://example.org/"

View File

@ -260,44 +260,6 @@ def test_context_managed_transport():
]
def test_context_managed_transport_and_mount():
class Transport(httpx.BaseTransport):
def __init__(self, name: str) -> None:
self.name: str = name
self.events: list[str] = []
def close(self):
# The base implementation of httpx.BaseTransport just
# calls into `.close`, so simple transport cases can just override
# this method for any cleanup, where more complex cases
# might want to additionally override `__enter__`/`__exit__`.
self.events.append(f"{self.name}.close")
def __enter__(self):
super().__enter__()
self.events.append(f"{self.name}.__enter__")
def __exit__(self, *args):
super().__exit__(*args)
self.events.append(f"{self.name}.__exit__")
transport = Transport(name="transport")
mounted = Transport(name="mounted")
with httpx.Client(transport=transport, mounts={"http://www.example.org": mounted}):
pass
assert transport.events == [
"transport.__enter__",
"transport.close",
"transport.__exit__",
]
assert mounted.events == [
"mounted.__enter__",
"mounted.close",
"mounted.__exit__",
]
def hello_world(request):
return httpx.Response(200, text="Hello, world!")
@ -364,41 +326,6 @@ def test_raw_client_header():
]
def unmounted(request: httpx.Request) -> httpx.Response:
data = {"app": "unmounted"}
return httpx.Response(200, json=data)
def mounted(request: httpx.Request) -> httpx.Response:
data = {"app": "mounted"}
return httpx.Response(200, json=data)
def test_mounted_transport():
transport = httpx.MockTransport(unmounted)
mounts = {"custom://": httpx.MockTransport(mounted)}
client = httpx.Client(transport=transport, mounts=mounts)
response = client.get("https://www.example.com")
assert response.status_code == 200
assert response.json() == {"app": "unmounted"}
response = client.get("custom://www.example.com")
assert response.status_code == 200
assert response.json() == {"app": "mounted"}
def test_all_mounted_transport():
mounts = {"all://": httpx.MockTransport(mounted)}
client = httpx.Client(mounts=mounts)
response = client.get("https://www.example.com")
assert response.status_code == 200
assert response.json() == {"app": "mounted"}
def test_server_extensions(server):
url = server.url.copy_with(path="/http_version_2")
with httpx.Client(http2=True) as client:

View File

@ -235,59 +235,3 @@ def test_host_with_non_default_port_in_url():
def test_request_auto_headers():
request = httpx.Request("GET", "https://www.example.org/")
assert "host" in request.headers
def test_same_origin():
origin = httpx.URL("https://example.com")
request = httpx.Request("GET", "HTTPS://EXAMPLE.COM:443")
client = httpx.Client()
headers = client._redirect_headers(request, origin, "GET")
assert headers["Host"] == request.url.netloc.decode("ascii")
def test_not_same_origin():
origin = httpx.URL("https://example.com")
request = httpx.Request("GET", "HTTP://EXAMPLE.COM:80")
client = httpx.Client()
headers = client._redirect_headers(request, origin, "GET")
assert headers["Host"] == origin.netloc.decode("ascii")
def test_is_https_redirect():
url = httpx.URL("https://example.com")
request = httpx.Request(
"GET", "http://example.com", headers={"Authorization": "empty"}
)
client = httpx.Client()
headers = client._redirect_headers(request, url, "GET")
assert "Authorization" in headers
def test_is_not_https_redirect():
url = httpx.URL("https://www.example.com")
request = httpx.Request(
"GET", "http://example.com", headers={"Authorization": "empty"}
)
client = httpx.Client()
headers = client._redirect_headers(request, url, "GET")
assert "Authorization" not in headers
def test_is_not_https_redirect_if_not_default_ports():
url = httpx.URL("https://example.com:1337")
request = httpx.Request(
"GET", "http://example.com:9999", headers={"Authorization": "empty"}
)
client = httpx.Client()
headers = client._redirect_headers(request, url, "GET")
assert "Authorization" not in headers

View File

@ -3,35 +3,35 @@ import httpx
def test_client_base_url():
client = httpx.Client()
client.base_url = "https://www.example.org/"
client.base_url = "https://www.example.org/" # type: ignore
assert isinstance(client.base_url, httpx.URL)
assert client.base_url == "https://www.example.org/"
def test_client_base_url_without_trailing_slash():
client = httpx.Client()
client.base_url = "https://www.example.org/path"
client.base_url = "https://www.example.org/path" # type: ignore
assert isinstance(client.base_url, httpx.URL)
assert client.base_url == "https://www.example.org/path/"
def test_client_base_url_with_trailing_slash():
client = httpx.Client()
client.base_url = "https://www.example.org/path/"
client.base_url = "https://www.example.org/path/" # type: ignore
assert isinstance(client.base_url, httpx.URL)
assert client.base_url == "https://www.example.org/path/"
def test_client_headers():
client = httpx.Client()
client.headers = {"a": "b"}
client.headers = {"a": "b"} # type: ignore
assert isinstance(client.headers, httpx.Headers)
assert client.headers["A"] == "b"
def test_client_cookies():
client = httpx.Client()
client.cookies = {"a": "b"}
client.cookies = {"a": "b"} # type: ignore
assert isinstance(client.cookies, httpx.Cookies)
mycookies = list(client.cookies.jar)
assert len(mycookies) == 1
@ -42,7 +42,7 @@ def test_client_timeout():
expected_timeout = 12.0
client = httpx.Client()
client.timeout = expected_timeout
client.timeout = expected_timeout # type: ignore
assert isinstance(client.timeout, httpx.Timeout)
assert client.timeout.connect == expected_timeout

View File

@ -1,265 +0,0 @@
import httpcore
import pytest
import httpx
def url_to_origin(url: str) -> httpcore.URL:
"""
Given a URL string, return the origin in the raw tuple format that
`httpcore` uses for it's representation.
"""
u = httpx.URL(url)
return httpcore.URL(scheme=u.raw_scheme, host=u.raw_host, port=u.port, target="/")
def test_socks_proxy():
url = httpx.URL("http://www.example.com")
for proxy in ("socks5://localhost/", "socks5h://localhost/"):
client = httpx.Client(proxy=proxy)
transport = client._transport_for_url(url)
assert isinstance(transport, httpx.HTTPTransport)
assert isinstance(transport._pool, httpcore.SOCKSProxy)
async_client = httpx.AsyncClient(proxy=proxy)
async_transport = async_client._transport_for_url(url)
assert isinstance(async_transport, httpx.AsyncHTTPTransport)
assert isinstance(async_transport._pool, httpcore.AsyncSOCKSProxy)
PROXY_URL = "http://[::1]"
@pytest.mark.parametrize(
["url", "proxies", "expected"],
[
("http://example.com", {}, None),
("http://example.com", {"https://": PROXY_URL}, None),
("http://example.com", {"http://example.net": PROXY_URL}, None),
# Using "*" should match any domain name.
("http://example.com", {"http://*": PROXY_URL}, PROXY_URL),
("https://example.com", {"http://*": PROXY_URL}, None),
# Using "example.com" should match example.com, but not www.example.com
("http://example.com", {"http://example.com": PROXY_URL}, PROXY_URL),
("http://www.example.com", {"http://example.com": PROXY_URL}, None),
# Using "*.example.com" should match www.example.com, but not example.com
("http://example.com", {"http://*.example.com": PROXY_URL}, None),
("http://www.example.com", {"http://*.example.com": PROXY_URL}, PROXY_URL),
# Using "*example.com" should match example.com and www.example.com
("http://example.com", {"http://*example.com": PROXY_URL}, PROXY_URL),
("http://www.example.com", {"http://*example.com": PROXY_URL}, PROXY_URL),
("http://wwwexample.com", {"http://*example.com": PROXY_URL}, None),
# ...
("http://example.com:443", {"http://example.com": PROXY_URL}, PROXY_URL),
("http://example.com", {"all://": PROXY_URL}, PROXY_URL),
("http://example.com", {"http://": PROXY_URL}, PROXY_URL),
("http://example.com", {"all://example.com": PROXY_URL}, PROXY_URL),
("http://example.com", {"http://example.com": PROXY_URL}, PROXY_URL),
("http://example.com", {"http://example.com:80": PROXY_URL}, PROXY_URL),
("http://example.com:8080", {"http://example.com:8080": PROXY_URL}, PROXY_URL),
("http://example.com:8080", {"http://example.com": PROXY_URL}, PROXY_URL),
(
"http://example.com",
{
"all://": PROXY_URL + ":1",
"http://": PROXY_URL + ":2",
"all://example.com": PROXY_URL + ":3",
"http://example.com": PROXY_URL + ":4",
},
PROXY_URL + ":4",
),
(
"http://example.com",
{
"all://": PROXY_URL + ":1",
"http://": PROXY_URL + ":2",
"all://example.com": PROXY_URL + ":3",
},
PROXY_URL + ":3",
),
(
"http://example.com",
{"all://": PROXY_URL + ":1", "http://": PROXY_URL + ":2"},
PROXY_URL + ":2",
),
],
)
def test_transport_for_request(url, proxies, expected):
mounts = {key: httpx.HTTPTransport(proxy=value) for key, value in proxies.items()}
client = httpx.Client(mounts=mounts)
transport = client._transport_for_url(httpx.URL(url))
if expected is None:
assert transport is client._transport
else:
assert isinstance(transport, httpx.HTTPTransport)
assert isinstance(transport._pool, httpcore.HTTPProxy)
assert transport._pool._proxy_url == url_to_origin(expected)
@pytest.mark.anyio
@pytest.mark.network
async def test_async_proxy_close():
try:
transport = httpx.AsyncHTTPTransport(proxy=PROXY_URL)
client = httpx.AsyncClient(mounts={"https://": transport})
await client.get("http://example.com")
finally:
await client.aclose()
@pytest.mark.network
def test_sync_proxy_close():
try:
transport = httpx.HTTPTransport(proxy=PROXY_URL)
client = httpx.Client(mounts={"https://": transport})
client.get("http://example.com")
finally:
client.close()
def test_unsupported_proxy_scheme():
with pytest.raises(ValueError):
httpx.Client(proxy="ftp://127.0.0.1")
@pytest.mark.parametrize(
["url", "env", "expected"],
[
("http://google.com", {}, None),
(
"http://google.com",
{"HTTP_PROXY": "http://example.com"},
"http://example.com",
),
# Auto prepend http scheme
("http://google.com", {"HTTP_PROXY": "example.com"}, "http://example.com"),
(
"http://google.com",
{"HTTP_PROXY": "http://example.com", "NO_PROXY": "google.com"},
None,
),
# Everything proxied when NO_PROXY is empty/unset
(
"http://127.0.0.1",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": ""},
"http://localhost:123",
),
# Not proxied if NO_PROXY matches URL.
(
"http://127.0.0.1",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "127.0.0.1"},
None,
),
# Proxied if NO_PROXY scheme does not match URL.
(
"http://127.0.0.1",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "https://127.0.0.1"},
"http://localhost:123",
),
# Proxied if NO_PROXY scheme does not match host.
(
"http://127.0.0.1",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "1.1.1.1"},
"http://localhost:123",
),
# Not proxied if NO_PROXY matches host domain suffix.
(
"http://courses.mit.edu",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "mit.edu"},
None,
),
# Proxied even though NO_PROXY matches host domain *prefix*.
(
"https://mit.edu.info",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "mit.edu"},
"http://localhost:123",
),
# Not proxied if one item in NO_PROXY case matches host domain suffix.
(
"https://mit.edu.info",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "mit.edu,edu.info"},
None,
),
# Not proxied if one item in NO_PROXY case matches host domain suffix.
# May include whitespace.
(
"https://mit.edu.info",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "mit.edu, edu.info"},
None,
),
# Proxied if no items in NO_PROXY match.
(
"https://mit.edu.info",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "mit.edu,mit.info"},
"http://localhost:123",
),
# Proxied if NO_PROXY domain doesn't match.
(
"https://foo.example.com",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "www.example.com"},
"http://localhost:123",
),
# Not proxied for subdomains matching NO_PROXY, with a leading ".".
(
"https://www.example1.com",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": ".example1.com"},
None,
),
# Proxied, because NO_PROXY subdomains only match if "." separated.
(
"https://www.example2.com",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "ample2.com"},
"http://localhost:123",
),
# No requests are proxied if NO_PROXY="*" is set.
(
"https://www.example3.com",
{"ALL_PROXY": "http://localhost:123", "NO_PROXY": "*"},
None,
),
],
)
@pytest.mark.parametrize("client_class", [httpx.Client, httpx.AsyncClient])
def test_proxies_environ(monkeypatch, client_class, url, env, expected):
for name, value in env.items():
monkeypatch.setenv(name, value)
client = client_class()
transport = client._transport_for_url(httpx.URL(url))
if expected is None:
assert transport == client._transport
else:
assert transport._pool._proxy_url == url_to_origin(expected)
@pytest.mark.parametrize(
["proxies", "is_valid"],
[
({"http": "http://127.0.0.1"}, False),
({"https": "http://127.0.0.1"}, False),
({"all": "http://127.0.0.1"}, False),
({"http://": "http://127.0.0.1"}, True),
({"https://": "http://127.0.0.1"}, True),
({"all://": "http://127.0.0.1"}, True),
],
)
def test_for_deprecated_proxy_params(proxies, is_valid):
mounts = {key: httpx.HTTPTransport(proxy=value) for key, value in proxies.items()}
if not is_valid:
with pytest.raises(ValueError):
httpx.Client(mounts=mounts)
else:
httpx.Client(mounts=mounts)
def test_proxy_with_mounts():
proxy_transport = httpx.HTTPTransport(proxy="http://127.0.0.1")
client = httpx.Client(mounts={"http://": proxy_transport})
transport = client._transport_for_url(httpx.URL("http://example.com"))
assert transport == proxy_transport

View File

@ -17,7 +17,7 @@ def test_client_queryparams_string():
assert client.params["a"] == "b"
client = httpx.Client()
client.params = "a=b"
client.params = "a=b" # type: ignore
assert isinstance(client.params, httpx.QueryParams)
assert client.params["a"] == "b"

View File

@ -174,46 +174,3 @@ def test_sensitive_headers(header):
value = "s3kr3t"
h = httpx.Headers({header: value})
assert repr(h) == "Headers({'%s': '[secure]'})" % header
@pytest.mark.parametrize(
"headers, output",
[
([("content-type", "text/html")], [("content-type", "text/html")]),
([("authorization", "s3kr3t")], [("authorization", "[secure]")]),
([("proxy-authorization", "s3kr3t")], [("proxy-authorization", "[secure]")]),
],
)
def test_obfuscate_sensitive_headers(headers, output):
as_dict = {k: v for k, v in output}
headers_class = httpx.Headers({k: v for k, v in headers})
assert repr(headers_class) == f"Headers({as_dict!r})"
@pytest.mark.parametrize(
"value, expected",
(
(
'<http:/.../front.jpeg>; rel=front; type="image/jpeg"',
[{"url": "http:/.../front.jpeg", "rel": "front", "type": "image/jpeg"}],
),
("<http:/.../front.jpeg>", [{"url": "http:/.../front.jpeg"}]),
("<http:/.../front.jpeg>;", [{"url": "http:/.../front.jpeg"}]),
(
'<http:/.../front.jpeg>; type="image/jpeg",<http://.../back.jpeg>;',
[
{"url": "http:/.../front.jpeg", "type": "image/jpeg"},
{"url": "http://.../back.jpeg"},
],
),
("", []),
),
)
def test_parse_header_links(value, expected):
all_links = httpx.Response(200, headers={"link": value}).links.values()
assert all(link in all_links for link in expected)
def test_parse_header_links_no_link():
all_links = httpx.Response(200).links
assert all_links == {}

View File

@ -226,16 +226,3 @@ def test_request_generator_content_picklable():
request.read()
pickle_request = pickle.loads(pickle.dumps(request))
assert pickle_request.content == b"test 123"
def test_request_params():
request = httpx.Request("GET", "http://example.com", params={})
assert str(request.url) == "http://example.com"
request = httpx.Request(
"GET", "http://example.com?c=3", params={"a": "1", "b": "2"}
)
assert str(request.url) == "http://example.com?a=1&b=2"
request = httpx.Request("GET", "http://example.com?a=1", params={})
assert str(request.url) == "http://example.com"

View File

@ -1011,10 +1011,7 @@ def test_response_decode_text_using_autodetect():
assert response.status_code == 200
assert response.reason_phrase == "OK"
# The encoded byte string is consistent with either ISO-8859-1 or
# WINDOWS-1252. Versions <6.0 of chardet claim the former, while chardet
# 6.0 detects the latter.
assert response.encoding in ("ISO-8859-1", "WINDOWS-1252")
assert response.encoding == "ISO-8859-1"
assert response.text == text

View File

@ -9,39 +9,39 @@ import httpx
def test_load_ssl_config():
context = httpx.create_ssl_context()
context = httpx.SSLContext()
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
assert context.check_hostname is True
def test_load_ssl_config_verify_non_existing_file():
with pytest.raises(IOError):
context = httpx.create_ssl_context()
context = httpx.SSLContext()
context.load_verify_locations(cafile="/path/to/nowhere")
def test_load_ssl_with_keylog(monkeypatch: typing.Any) -> None:
monkeypatch.setenv("SSLKEYLOGFILE", "test")
context = httpx.create_ssl_context()
context = httpx.SSLContext()
assert context.keylog_filename == "test"
def test_load_ssl_config_verify_existing_file():
context = httpx.create_ssl_context()
context = httpx.SSLContext()
context.load_verify_locations(capath=certifi.where())
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
assert context.check_hostname is True
def test_load_ssl_config_verify_directory():
context = httpx.create_ssl_context()
context = httpx.SSLContext()
context.load_verify_locations(capath=Path(certifi.where()).parent)
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
assert context.check_hostname is True
def test_load_ssl_config_cert_and_key(cert_pem_file, cert_private_key_file):
context = httpx.create_ssl_context()
context = httpx.SSLContext()
context.load_cert_chain(cert_pem_file, cert_private_key_file)
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
assert context.check_hostname is True
@ -51,7 +51,7 @@ def test_load_ssl_config_cert_and_key(cert_pem_file, cert_private_key_file):
def test_load_ssl_config_cert_and_encrypted_key(
cert_pem_file, cert_encrypted_private_key_file, password
):
context = httpx.create_ssl_context()
context = httpx.SSLContext()
context.load_cert_chain(cert_pem_file, cert_encrypted_private_key_file, password)
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
assert context.check_hostname is True
@ -61,7 +61,7 @@ def test_load_ssl_config_cert_and_key_invalid_password(
cert_pem_file, cert_encrypted_private_key_file
):
with pytest.raises(ssl.SSLError):
context = httpx.create_ssl_context()
context = httpx.SSLContext()
context.load_cert_chain(
cert_pem_file, cert_encrypted_private_key_file, "password1"
)
@ -69,23 +69,29 @@ def test_load_ssl_config_cert_and_key_invalid_password(
def test_load_ssl_config_cert_without_key_raises(cert_pem_file):
with pytest.raises(ssl.SSLError):
context = httpx.create_ssl_context()
context = httpx.SSLContext()
context.load_cert_chain(cert_pem_file)
def test_load_ssl_config_no_verify():
context = httpx.create_ssl_context(verify=False)
context = httpx.SSLContext(verify=False)
assert context.verify_mode == ssl.VerifyMode.CERT_NONE
assert context.check_hostname is False
def test_SSLContext_with_get_request(server, cert_pem_file):
context = httpx.create_ssl_context()
context = httpx.SSLContext()
context.load_verify_locations(cert_pem_file)
response = httpx.get(server.url, verify=context)
response = httpx.get(server.url, ssl_context=context)
assert response.status_code == 200
def test_SSLContext_repr():
ssl_context = httpx.SSLContext()
assert repr(ssl_context) == "<SSLContext(verify=True)>"
def test_limits_repr():
limits = httpx.Limits(max_connections=100)
expected = (
@ -182,3 +188,18 @@ def test_proxy_with_auth_from_url():
def test_invalid_proxy_scheme():
with pytest.raises(ValueError):
httpx.Proxy("invalid://example.com")
def test_certifi_lazy_loading():
global httpx, certifi
import sys
del sys.modules["httpx"]
del sys.modules["certifi"]
del httpx
del certifi
import httpx
assert "certifi" not in sys.modules
_context = httpx.SSLContext()
assert "certifi" in sys.modules

View File

@ -4,6 +4,7 @@ import typing
import pytest
import httpx
from httpx._content import encode_json
method = "POST"
url = "https://www.example.com"
@ -488,20 +489,24 @@ def test_response_invalid_argument():
def test_ensure_ascii_false_with_french_characters():
data = {"greeting": "Bonjour, ça va ?"}
response = httpx.Response(200, json=data)
assert "ça va" in response.text, (
"ensure_ascii=False should preserve French accented characters"
)
assert response.headers["Content-Type"] == "application/json"
headers, byte_stream = encode_json(data)
json_output = b"".join(byte_stream).decode("utf-8")
assert (
"ça va" in json_output
), "ensure_ascii=False should preserve French accented characters"
assert headers["Content-Type"] == "application/json"
def test_separators_for_compact_json():
data = {"clé": "valeur", "liste": [1, 2, 3]}
response = httpx.Response(200, json=data)
assert response.text == '{"clé":"valeur","liste":[1,2,3]}', (
"separators=(',', ':') should produce a compact representation"
)
assert response.headers["Content-Type"] == "application/json"
headers, byte_stream = encode_json(data)
json_output = b"".join(byte_stream).decode("utf-8")
assert (
json_output == '{"clé":"valeur","liste":[1,2,3]}'
), "separators=(',', ':') should produce a compact representation"
assert headers["Content-Type"] == "application/json"
def test_allow_nan_false():
@ -511,8 +516,8 @@ def test_allow_nan_false():
with pytest.raises(
ValueError, match="Out of range float values are not JSON compliant"
):
httpx.Response(200, json=data_with_nan)
encode_json(data_with_nan)
with pytest.raises(
ValueError, match="Out of range float values are not JSON compliant"
):
httpx.Response(200, json=data_with_inf)
encode_json(data_with_inf)

View File

@ -100,25 +100,6 @@ def test_zstd_decoding_error():
)
def test_zstd_empty():
headers = [(b"Content-Encoding", b"zstd")]
response = httpx.Response(200, headers=headers, content=b"")
assert response.content == b""
def test_zstd_truncated():
body = b"test 123"
compressed_body = zstd.compress(body)
headers = [(b"Content-Encoding", b"zstd")]
with pytest.raises(httpx.DecodingError):
httpx.Response(
200,
headers=headers,
content=compressed_body[1:3],
)
def test_zstd_multiframe():
# test inspired by urllib3 test suite
data = (

View File

@ -1,12 +1,9 @@
import json
import logging
import os
import random
import pytest
import httpx
from httpx._utils import URLPattern, get_environment_proxies
@pytest.mark.parametrize(
@ -50,6 +47,35 @@ def test_guess_by_bom(encoding, expected):
assert response.json() == {"abc": 123}
@pytest.mark.parametrize(
"value, expected",
(
(
'<http:/.../front.jpeg>; rel=front; type="image/jpeg"',
[{"url": "http:/.../front.jpeg", "rel": "front", "type": "image/jpeg"}],
),
("<http:/.../front.jpeg>", [{"url": "http:/.../front.jpeg"}]),
("<http:/.../front.jpeg>;", [{"url": "http:/.../front.jpeg"}]),
(
'<http:/.../front.jpeg>; type="image/jpeg",<http://.../back.jpeg>;',
[
{"url": "http:/.../front.jpeg", "type": "image/jpeg"},
{"url": "http://.../back.jpeg"},
],
),
("", []),
),
)
def test_parse_header_links(value, expected):
all_links = httpx.Response(200, headers={"link": value}).links.values()
assert all(link in all_links for link in expected)
def test_parse_header_links_no_link():
all_links = httpx.Response(200).links
assert all_links == {}
def test_logging_request(server, caplog):
caplog.set_level(logging.INFO)
with httpx.Client() as client:
@ -87,64 +113,70 @@ def test_logging_redirect_chain(server, caplog):
@pytest.mark.parametrize(
["environment", "proxies"],
"headers, output",
[
({}, {}),
({"HTTP_PROXY": "http://127.0.0.1"}, {"http://": "http://127.0.0.1"}),
(
{"https_proxy": "http://127.0.0.1", "HTTP_PROXY": "https://127.0.0.1"},
{"https://": "http://127.0.0.1", "http://": "https://127.0.0.1"},
),
({"all_proxy": "http://127.0.0.1"}, {"all://": "http://127.0.0.1"}),
({"TRAVIS_APT_PROXY": "http://127.0.0.1"}, {}),
({"no_proxy": "127.0.0.1"}, {"all://127.0.0.1": None}),
({"no_proxy": "192.168.0.0/16"}, {"all://192.168.0.0/16": None}),
({"no_proxy": "::1"}, {"all://[::1]": None}),
({"no_proxy": "localhost"}, {"all://localhost": None}),
({"no_proxy": "github.com"}, {"all://*github.com": None}),
({"no_proxy": ".github.com"}, {"all://*.github.com": None}),
({"no_proxy": "http://github.com"}, {"http://github.com": None}),
([("content-type", "text/html")], [("content-type", "text/html")]),
([("authorization", "s3kr3t")], [("authorization", "[secure]")]),
([("proxy-authorization", "s3kr3t")], [("proxy-authorization", "[secure]")]),
],
)
def test_get_environment_proxies(environment, proxies):
os.environ.update(environment)
assert get_environment_proxies() == proxies
def test_obfuscate_sensitive_headers(headers, output):
as_dict = {k: v for k, v in output}
headers_class = httpx.Headers({k: v for k, v in headers})
assert repr(headers_class) == f"Headers({as_dict!r})"
@pytest.mark.parametrize(
["pattern", "url", "expected"],
[
("http://example.com", "http://example.com", True),
("http://example.com", "https://example.com", False),
("http://example.com", "http://other.com", False),
("http://example.com:123", "http://example.com:123", True),
("http://example.com:123", "http://example.com:456", False),
("http://example.com:123", "http://example.com", False),
("all://example.com", "http://example.com", True),
("all://example.com", "https://example.com", True),
("http://", "http://example.com", True),
("http://", "https://example.com", False),
("all://", "https://example.com:123", True),
("", "https://example.com:123", True),
],
)
def test_url_matches(pattern, url, expected):
pattern = URLPattern(pattern)
assert pattern.matches(httpx.URL(url)) == expected
def test_same_origin():
origin = httpx.URL("https://example.com")
request = httpx.Request("GET", "HTTPS://EXAMPLE.COM:443")
client = httpx.Client()
headers = client._redirect_headers(request, origin, "GET")
assert headers["Host"] == request.url.netloc.decode("ascii")
def test_pattern_priority():
matchers = [
URLPattern("all://"),
URLPattern("http://"),
URLPattern("http://example.com"),
URLPattern("http://example.com:123"),
]
random.shuffle(matchers)
assert sorted(matchers) == [
URLPattern("http://example.com:123"),
URLPattern("http://example.com"),
URLPattern("http://"),
URLPattern("all://"),
]
def test_not_same_origin():
origin = httpx.URL("https://example.com")
request = httpx.Request("GET", "HTTP://EXAMPLE.COM:80")
client = httpx.Client()
headers = client._redirect_headers(request, origin, "GET")
assert headers["Host"] == origin.netloc.decode("ascii")
def test_is_https_redirect():
url = httpx.URL("https://example.com")
request = httpx.Request(
"GET", "http://example.com", headers={"Authorization": "empty"}
)
client = httpx.Client()
headers = client._redirect_headers(request, url, "GET")
assert "Authorization" in headers
def test_is_not_https_redirect():
url = httpx.URL("https://www.example.com")
request = httpx.Request(
"GET", "http://example.com", headers={"Authorization": "empty"}
)
client = httpx.Client()
headers = client._redirect_headers(request, url, "GET")
assert "Authorization" not in headers
def test_is_not_https_redirect_if_not_default_ports():
url = httpx.URL("https://example.com:1337")
request = httpx.Request(
"GET", "http://example.com:9999", headers={"Authorization": "empty"}
)
client = httpx.Client()
headers = client._redirect_headers(request, url, "GET")
assert "Authorization" not in headers