Niquests Documentation – Drop-in replacement for Requests – HTTP/1.1, HTTP/2, and HTTP/3.

Make a request¶

Making a request with Niquests is very simple.

Begin by importing the Niquests module:

Now, let’s try to get a webpage. For this example, let’s get GitHub’s public timeline.

r = niquests.get('https://api.github.com/events')

r = await niquests.aget('https://api.github.com/events')

Now, we have a Response object called r. We can get all the information we need from this object.

Niquests’ simple API makes all forms of HTTP requests straightforward. For example, this is how you make an HTTP POST request:

r = niquests.post('https://httpbingo.org/post', data={'key': 'value'})

r = await niquests.apost('https://httpbingo.org/post', data={'key': 'value'})

Nice, right? What about the other HTTP request methods: PUT, PATCH, DELETE, HEAD, OPTIONS, and QUERY? These are all just as simple:

r = niquests.put('https://httpbingo.org/put', data={'key': 'value'})
r = niquests.patch('https://httpbingo.org/patch', data={'key': 'value'})
r = niquests.delete('https://httpbingo.org/delete')
r = niquests.head('https://httpbingo.org/get')
r = niquests.options('https://httpbingo.org/get')
r = niquests.query('https://httpbingo.org/anything', json={'key': 'value'})

r = await niquests.aput('https://httpbingo.org/put', data={'key': 'value'})
r = await niquests.apatch('https://httpbingo.org/patch', data={'key': 'value'})
r = await niquests.adelete('https://httpbingo.org/delete')
r = await niquests.ahead('https://httpbingo.org/get')
r = await niquests.aoptions('https://httpbingo.org/get')
r = await niquests.aquery('https://httpbingo.org/anything', json={'key': 'value'})

QUERY, standardized by RFC 10008, carries a query in the request body. Like GET, it is safe and idempotent; unlike GET, it can carry content. Use Session.query() or AsyncSession.query() when working with a session.

That’s all well and good, but it’s also only the start of what Niquests can do.

Passing parameters in URLs¶

You often want to send some sort of data in the URL’s query string. If you were constructing the URL by hand, this data would be given as key/value pairs in the URL after a question mark, e.g. httpbingo.org/get?key=val. Niquests allows you to provide these arguments as a dictionary of strings, using the params keyword argument. As an example, if you wanted to pass key1=value1 and key2=value2 to httpbingo.org/get, you would use the following code:

payload = {'key1': 'value1', 'key2': 'value2'}
r = niquests.get('https://httpbingo.org/get', params=payload)

payload = {'key1': 'value1', 'key2': 'value2'}
r = await niquests.aget('https://httpbingo.org/get', params=payload)

You can see that the URL has been correctly encoded by printing the URL:

print(r.url)  # 'https://httpbingo.org/get?key1=value1&key2=value2'

Note that any dictionary key whose value is None will not be added to the URL’s query string.

You can also pass a list of items as a value:

payload = {'key1': 'value1', 'key2': ['value2', 'value3']}
r = niquests.get('https://httpbingo.org/get', params=payload)

print(r.url)  # 'https://httpbingo.org/get?key1=value1&key2=value2&key2=value3'

payload = {'key1': 'value1', 'key2': ['value2', 'value3']}
r = await niquests.aget('https://httpbingo.org/get', params=payload)

print(r.url)  # 'https://httpbingo.org/get?key1=value1&key2=value2&key2=value3'

Response content¶

We can read the content of the server’s response. Consider the GitHub timeline again:

import niquests

r = niquests.get('https://api.github.com/events')
print(r.text)  # '[{"repository":{"open_issues":0,"url":"https://github.com/...

import niquests

r = await niquests.aget('https://api.github.com/events')
print(r.text)  # '[{"repository":{"open_issues":0,"url":"https://github.com/...

Niquests automatically decodes content from the server. Most Unicode character sets are decoded seamlessly.

When you make a request, Niquests makes educated guesses about the encoding of the response based on the HTTP headers. The text encoding guessed by Niquests is used when you access r.text. You can inspect and change it through the r.encoding property:

print(r.encoding)  # 'utf-8'

r.encoding = 'ISO-8859-1'  # Force a specific encoding.

If you change the encoding, Niquests will use the new value of r.encoding whenever you access r.text. You might do this when you can apply special logic to work out what the encoding of the content will be. For example, HTML and XML can specify their encoding in their bodies. In situations like this, use r.content to find the encoding, and then set r.encoding. This will let you use r.text with the correct encoding.

Niquests will also use custom encodings if you need them. If you have created your own encoding and registered it with the codecs module, you can simply use the codec name as the value of r.encoding and Niquests will handle the decoding for you.

Binary response content¶

You can also access the response body as bytes, for non-text requests:

>>> r.content
b'[{"repository":{"open_issues":0,"url":"https://github.com/...

The gzip and deflate content codings are automatically decoded for you.

The br content coding is automatically decoded for you if a Brotli library like brotli or brotlicffi is installed.

The zstd content coding is automatically decoded on Python 3.14 and later using the standard-library compression.zstd module. On older Python versions, install the zstandard library or the niquests[zstd] extra to enable it.

For example, to create an image from binary data returned by a request, you can use the following code:

>>> from PIL import Image
>>> from io import BytesIO

>>> i = Image.open(BytesIO(r.content))

JSON response content¶

There’s also a built-in JSON decoder for JSON data:

import niquests

r = niquests.get('https://api.github.com/events')
print(r.json())  # [{'repository': {'open_issues': 0, 'url': 'https://github.com/...

import niquests

r = await niquests.aget('https://api.github.com/events')
print(r.json())  # [{'repository': {'open_issues': 0, 'url': 'https://github.com/...

In case the JSON decoding fails, r.json() raises an exception. For example, if the response gets a 204 (No Content), or if the response contains invalid JSON, attempting r.json() raises JSONDecodeError. This wrapper exception provides interoperability for multiple exceptions that may be thrown by different Python versions and JSON serialization libraries. Response.json() attempts to parse the response body regardless of its Content-Type header.

The success of a call to r.json() does not indicate the success of the response. Some servers may return a JSON object in a failed response (e.g. error details with HTTP 500). Such JSON will be decoded and returned. To check that a request is successful, use r.raise_for_status() or check that r.status_code is what you expect.

Raw response content¶

In the rare case that you’d like to get the raw socket response from the server, you can access r.raw. If you want to do this, make sure you set stream=True in your initial request. Once you do, you can do this:

r = niquests.get('https://api.github.com/events', stream=True)

r.raw
# <urllib3.response.HTTPResponse object at ...>

r.raw.read(10)
# b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x03'

r = await niquests.aget('https://api.github.com/events', stream=True)

r.raw
# <urllib3._async.response.AsyncHTTPResponse object at ...>

await r.raw.read(10)
# b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x03'

In general, however, you should use a pattern like this to save what is being streamed to a file:

with open(filename, 'wb') as fd:
    for chunk in r.iter_content(chunk_size=128):
        fd.write(chunk)

with open(filename, 'wb') as fd:
    async for chunk in await r.iter_content(chunk_size=128):
        fd.write(chunk)

Using Response.iter_content will handle a lot of what you would otherwise have to handle when using Response.raw directly. When streaming a download, the above is the preferred and recommended way to retrieve the content. Note that chunk_size can be freely adjusted to a number that may better fit your use cases.

More complicated POST requests¶

Typically, you want to send form-encoded data, much like an HTML form. To do this, simply pass a dictionary to the data argument. Your dictionary of data will automatically be form-encoded when the request is made:

payload = {'key1': 'value1', 'key2': 'value2'}
r = niquests.post('https://httpbingo.org/post', data=payload)
print(r.json()['form'])  # {'key1': ['value1'], 'key2': ['value2']}

payload = {'key1': 'value1', 'key2': 'value2'}
r = await niquests.apost('https://httpbingo.org/post', data=payload)
print(r.json()['form'])  # {'key1': ['value1'], 'key2': ['value2']}

The data argument can also have multiple values for each key. This can be done by making data either a list of tuples or a dictionary with lists as values. This is particularly useful when the form has multiple elements that use the same key:

payload_tuples = [('key1', 'value1'), ('key1', 'value2')]
r1 = niquests.post('https://httpbingo.org/post', data=payload_tuples)
payload_dict = {'key1': ['value1', 'value2']}
r2 = niquests.post('https://httpbingo.org/post', data=payload_dict)
assert r1.json()['form'] == r2.json()['form']

payload_tuples = [('key1', 'value1'), ('key1', 'value2')]
r1 = await niquests.apost('https://httpbingo.org/post', data=payload_tuples)
payload_dict = {'key1': ['value1', 'value2']}
r2 = await niquests.apost('https://httpbingo.org/post', data=payload_dict)
assert r1.json()['form'] == r2.json()['form']

There are times that you may want to send data that is not form-encoded. If you pass in a string instead of a dictionary, that data will be posted directly.

For example, to encode JSON manually:

import json

url = 'https://httpbingo.org/post'
payload = {'some': 'data'}
r = niquests.post(url, data=json.dumps(payload))

import json

url = 'https://httpbingo.org/post'
payload = {'some': 'data'}
r = await niquests.apost(url, data=json.dumps(payload))

The code above does not add a Content-Type: application/json header.

If you need that header set and do not want to encode the dictionary yourself, you can pass the object directly using the json parameter, and it will be encoded automatically:

r = niquests.post(url, json=payload)

r = await niquests.apost(url, json=payload)

The json parameter is ignored if non-empty data or files is passed.

import msgspec
import niquests

class User(msgspec.Struct):
    name: str
    active: bool

payload = User(name="Lina", active=True)
encoder = msgspec.json.encode
url = "https://httpbingo.org/post"
r = niquests.post(url, json=payload, json_encoder=encoder)

with niquests.Session(json_encoder=encoder) as session:
    r = session.post(url, json=payload)

import msgspec
import niquests

class User(msgspec.Struct):
    name: str
    active: bool

payload = User(name="Lina", active=True)
encoder = msgspec.json.encode
url = "https://httpbingo.org/post"
r = await niquests.apost(url, json=payload, json_encoder=encoder)

async with niquests.AsyncSession(json_encoder=encoder) as session:
    r = await session.post(url, json=payload)

POST multipart form data without a file¶

Since Niquests 3.1.2, you can override the default application/x-www-form-urlencoded encoding and submit multipart form data without a file:

r = niquests.post(
    url, data=payload, headers={'Content-Type': 'multipart/form-data'}
)

r = await niquests.apost(
    url, data=payload, headers={'Content-Type': 'multipart/form-data'}
)

POST a multipart-encoded file¶

Niquests makes it simple to upload multipart-encoded files:

with open('report.xls', 'rb') as report:
    r = niquests.post(url, files={'file': report})

with open('report.xls', 'rb') as report:
    r = await niquests.apost(url, files={'file': report})

You can set the filename, content type, and headers explicitly:

with open('report.xls', 'rb') as report:
    files = {
        'file': (
            'report.xls', report, 'application/vnd.ms-excel', {'Expires': '0'}
        )
    }
    r = niquests.post(url, files=files)

with open('report.xls', 'rb') as report:
    files = {
        'file': (
            'report.xls', report, 'application/vnd.ms-excel', {'Expires': '0'}
        )
    }
    r = await niquests.apost(url, files=files)

You can also send strings as files:

files = {'file': ('report.csv', 'some,data,to,send\nanother,row,to,send\n')}
r = niquests.post(url, files=files)

files = {'file': ('report.csv', 'some,data,to,send\nanother,row,to,send\n')}
r = await niquests.apost(url, files=files)

In the event you are posting a very large file as a multipart/form-data request, you may want to stream the request. By default, Niquests does not provide a multipart streaming encoder, but a separate package does: requests-toolbelt. You should read the toolbelt’s documentation for more details about how to use it.

For information about sending multiple files in one request, see the advanced section.

Response status codes¶

We can check the response status code:

>>> r = niquests.get('https://httpbingo.org/get')
>>> r.status_code
200

Niquests also comes with a built-in status code lookup object for easy reference:

>>> r.status_code == niquests.codes.ok
True

If a request returns a 4xx client error or 5xx server error response, we can raise an exception with Response.raise_for_status():

>>> bad_r = niquests.get('https://httpbingo.org/status/404')
>>> bad_r.status_code
404

>>> try:
...     bad_r.raise_for_status()
... except niquests.exceptions.HTTPError:
...     print('The response was unsuccessful')
The response was unsuccessful

Because the status_code for r is 200, raise_for_status returns the response itself:

>>> r.raise_for_status() is r
True

All is well.

Cookies¶

If a response contains cookies, you can quickly access them:

>>> url = 'https://httpbingo.org/cookies/set?example_cookie_name=example_cookie_value'
>>> r = niquests.get(url, allow_redirects=False)

>>> r.cookies['example_cookie_name']
'example_cookie_value'

To send your own cookies to the server, you can use the cookies parameter:

>>> url = 'https://httpbingo.org/cookies'
>>> cookies = dict(cookies_are='working')

>>> r = niquests.get(url, cookies=cookies)
>>> r.json()['cookies']
{'cookies_are': 'working'}

Cookies are returned in a RequestsCookieJar, which acts like a dictionary but also offers a more complete interface, suitable for use over multiple domains or paths. Cookie jars can also be passed in to requests:

>>> jar = niquests.cookies.RequestsCookieJar()
>>> jar.set('tasty_cookie', 'yum', domain='httpbingo.org', path='/cookies')
>>> jar.set('gross_cookie', 'blech', domain='httpbingo.org', path='/elsewhere')
>>> url = 'https://httpbingo.org/cookies'
>>> r = niquests.get(url, cookies=jar)
>>> r.json()['cookies']
{'tasty_cookie': 'yum'}

The same response and request cookie APIs are available asynchronously:

r = await niquests.aget(url, cookies=cookies)
print(r.cookies)

session.cookies = MozillaCookieJar("cookies.txt")  # type: ignore[assignment]

By default, sessions merge cookies received through Set-Cookie into their cookie jar. Set allow_incoming_cookies=False to prevent that merge. Response cookies remain available, and cookies you set explicitly are still sent:

with niquests.Session(allow_incoming_cookies=False) as session:
    session.cookies.set('outgoing', 'yes')
    r = session.get('https://httpbingo.org/cookies/set?incoming=no')
    assert 'incoming' not in session.cookies

async with niquests.AsyncSession(allow_incoming_cookies=False) as session:
    session.cookies.set('outgoing', 'yes')
    r = await session.get('https://httpbingo.org/cookies/set?incoming=no')
    assert 'incoming' not in session.cookies

Redirection and history¶

By default, Niquests follows redirects for all methods except HEAD.

We can use the history property of the response object to track redirects.

The Response.history list contains the Response objects that were created in order to complete the request. The list is sorted from the oldest to the most recent response.

For example, GitHub redirects all HTTP requests to HTTPS:

>>> r = niquests.get('http://github.com/')
>>> r.url
'https://github.com/'
>>> r.status_code
200
>>> r.history
[<Response HTTP/2 [301]>]

If you’re using GET, OPTIONS, POST, PUT, PATCH, DELETE, or QUERY, you can disable redirection handling with the allow_redirects parameter:

>>> r = niquests.get('http://github.com/', allow_redirects=False)
>>> r.status_code
301
>>> r.history
[]

If you’re using HEAD, you can enable redirection as well:

>>> r = niquests.head('http://github.com/', allow_redirects=True)
>>> r.url
'https://github.com/'
>>> r.history
[<Response HTTP/2 [301]>]

The redirect controls and history property are identical in async code:

r = await niquests.aget('http://github.com/', allow_redirects=False)
assert r.status_code == 301
assert r.history == []

r = await niquests.ahead('http://github.com/', allow_redirects=True)
assert r.url == 'https://github.com/'
assert len(r.history) == 1

Timeouts¶

You can limit how long Niquests waits during network operations with the timeout parameter. Nearly all production requests should specify a timeout:

try:
    niquests.get('https://github.com/', timeout=0.001)
except niquests.exceptions.Timeout:
    print('The request timed out')

try:
    await niquests.aget('https://github.com/', timeout=0.001)
except niquests.exceptions.Timeout:
    print('The request timed out')

Errors and exceptions¶

In the event of a network problem, such as a DNS failure or refused connection, Niquests will raise a ConnectionError exception.

Response.raise_for_status() will raise an HTTPError if the HTTP request returned an unsuccessful status code.

If a request times out, a Timeout exception is raised.

If a request exceeds the configured maximum number of redirects, a TooManyRedirects exception is raised.

All exceptions that Niquests explicitly raises inherit from niquests.exceptions.RequestException.

HTTP/3 over QUIC¶

Niquests relies on urllib3.future and the semi-optional qh3 package for HTTP/3. If qh3 is not installed, HTTP/3 and QUIC are unavailable, but HTTP/1.1 and HTTP/2 continue to work. Installing qh3 may require a compilation toolchain.

Run python -m niquests.help to check whether the installed dependencies support HTTP/3. You can inspect the protocol negotiated for a response:

r = niquests.get('https://1.1.1.1')
print(r.http_version)

r = await niquests.aget('https://1.1.1.1')
print(r.http_version)

The underlying library understands the Alt-Svc header and looks for an h3 alternative service. Once a valid service is discovered, Niquests can open a QUIC connection and caches that information in memory. Negotiation depends on the peer, DNS and Alt-Svc information, cached state, network conditions, and installed dependencies. Repeated calls do not guarantee a particular HTTP/2-to-HTTP/3 sequence.

Lazy responses and manual scheduling¶

HTTP/2 and HTTP/3 multiplexing is automatic in Niquests and does not require this option. The historically named multiplexed=True option instead enables manual response scheduling.

In this mode, each request is submitted immediately, but its request method returns a lazy, promise-backed response before waiting for the exchange to complete. This lets you submit several requests before resolving any of their responses.

When the peer supports HTTP/2 or HTTP/3, the outstanding exchanges can progress concurrently as independent streams on the same connection. Leaving one response unresolved does not prevent later requests from using that connection.

To benefit from this mode, submit multiple requests before accessing response data. Resolve the resulting promises explicitly with Session.gather() or through the other resolution mechanisms described below.

from niquests import Session
with Session(multiplexed=True) as s:
    responses = [
        s.get("https://httpbingo.org/delay/3"),
        s.get("https://httpbingo.org/delay/1"),
    ]
    assert all(response.lazy for response in responses)

    s.gather()  # Resolve pending responses before closing the session.

from niquests import AsyncSession
async with AsyncSession(multiplexed=True) as s:
    responses = [
        await s.get("https://httpbingo.org/delay/3"),
        await s.get("https://httpbingo.org/delay/1"),
    ]
    assert all(response.lazy for response in responses)

    await s.gather()  # Resolve pending responses before closing the session.

from niquests import Session

with Session(multiplexed=True) as s:
    responses = [
        s.get("https://httpbingo.org/delay/3"),
        s.get("https://httpbingo.org/delay/1"),
    ]
    s.gather(responses[0])
    print(responses[0].status_code)
    assert responses[1].lazy is True

    s.gather(responses[1])

from niquests import AsyncSession

async with AsyncSession(multiplexed=True) as s:
    responses = [
        await s.get("https://httpbingo.org/delay/3"),
        await s.get("https://httpbingo.org/delay/1"),
    ]
    await s.gather(responses[0])
    print(responses[0].status_code)
    assert responses[1].lazy is True

    await s.gather(responses[1])

with Session(multiplexed=True) as s:
    response = s.get("https://httpbingo.org/delay/1")
    print(response.status_code)  # Resolves this response first.

s.gather()  # Resolve all pending responses.
s.gather(resp)  # Resolve only resp.
s.gather(max_fetch=2)  # Resolve up to two available responses per adapter.
s.gather(resp_a, resp_b, resp_c)  # Resolve these three responses.
s.gather(resp_a, resp_b, resp_c, max_fetch=1)  # Resolve one available response per adapter.

await s.gather()  # Resolve all pending responses.
await s.gather(resp)  # Resolve only resp.
await s.gather(max_fetch=2)  # Resolve up to two available responses per adapter.
await s.gather(resp_a, resp_b, resp_c)  # Resolve these three responses.
await s.gather(resp_a, resp_b, resp_c, max_fetch=1)  # Resolve one available response per adapter.

Async session¶

Niquests provides AsyncSession for awaitable HTTP requests. Its request methods mirror Session and return coroutines.

Here is a basic example:

import asyncio
from niquests import AsyncSession, Response

async def main() -> None:
    async with AsyncSession() as s:
        tasks = [s.get("https://httpbingo.org/delay/1") for _ in range(10)]
        responses = await asyncio.gather(*tasks)
        print(responses)

if __name__ == "__main__":
    asyncio.run(main())

Async lazy responses and manual scheduling¶

AsyncSession(multiplexed=True) uses the same manual response scheduling described above. It does not enable HTTP/2 or HTTP/3 multiplexing; protocol negotiation remains automatic.

Each awaited request method submits its request and returns a lazy, promise-backed response without waiting for that exchange to finish. Submit several requests before resolving them so HTTP/2 or HTTP/3 streams can progress concurrently on the underlying connection. Resolve the promises explicitly with await session.gather().

Look at this basic sample:

import asyncio
from niquests import AsyncSession

async def main() -> None:
    async with AsyncSession(multiplexed=True) as s:
        responses = [
            await s.get("https://httpbingo.org/delay/1") for _ in range(10)
        ]
        assert all(response.lazy for response in responses)

        await s.gather()
        assert all(response.lazy is False for response in responses)
        print(responses)

if __name__ == "__main__":
    asyncio.run(main())

Unlike synchronous lazy responses, an asynchronous lazy response cannot resolve itself while a non-awaitable attribute is being accessed, because doing so would block the event loop. Gather it explicitly before that access.

AsyncResponse for streams¶

Delaying the content consumption in an async context can be easily achieved using:

import niquests
import asyncio

async def main() -> None:

    async with niquests.AsyncSession() as s:
        r = await s.get("https://httpbingo.org/get", stream=True)

        async for chunk in await r.iter_content(16):
            print(chunk)

if __name__ == "__main__":

    asyncio.run(main())

Or use iter_lines. It is an async generator, so iterate over it directly without await:

import niquests
import asyncio

async def main() -> None:

    async with niquests.AsyncSession() as s:
        r = await s.get("https://httpbingo.org/get", stream=True)

        async for line in r.iter_lines():
            print(line)

if __name__ == "__main__":
    asyncio.run(main())

Or simply by doing:

import niquests
import asyncio

async def main() -> None:

    async with niquests.AsyncSession() as s:
        r = await s.get("https://httpbingo.org/get", stream=True)
        payload = await r.json()

if __name__ == "__main__":

    asyncio.run(main())

When you specify stream=True with AsyncSession, the returned object is an AsyncResponse. Its iter_content and iter_raw methods are awaitable and return async iterators, so use async for chunk in await response.iter_content(). In contrast, iter_lines is an async generator and is used as async for line in response.iter_lines() without await. The content, json, text, and close interfaces are also awaitable.

When enabling multiplexing in an async context, call await s.gather() before direct access to non-awaitable response interfaces.

Here is a basic example of how you would do it:

import niquests
import asyncio

async def main() -> None:

    responses = []

    async with niquests.AsyncSession(multiplexed=True) as s:
        responses.append(
            await s.get("https://httpbingo.org/get", stream=True)
        )
        responses.append(
            await s.get("https://httpbingo.org/get", stream=True)
        )

        print(responses)

        await s.gather()

        print(responses)

        for response in responses:
            async for chunk in await response.iter_content(16):
                print(chunk)


if __name__ == "__main__":

    asyncio.run(main())

Scale your Session / pool¶

By default, Niquests retains up to 10 origin pools and configures each pool with a capacity of 10 connections. You can increase or decrease these values.

Set the following parameters in a session constructor:

Session(pool_connections=10, pool_maxsize=10)

• pool_connections is the number of origin connection pools retained in the cache.

• pool_maxsize is the configured connection capacity of each origin pool.

import niquests

with niquests.Session(pool_connections=2) as s:
    s.get("https://host-a.tld/some")
    s.get("https://host-b.tld/some")
    s.get("https://host-c.tld/some")

import niquests

async with niquests.AsyncSession(pool_connections=2) as s:
    await s.get("https://host-a.tld/some")
    await s.get("https://host-b.tld/some")
    await s.get("https://host-c.tld/some")

DNS resolution¶

Niquests has built-in support for DNS over HTTPS, DNS over TLS, DNS over UDP, and DNS over QUIC. Encrypted resolvers use the configured trust store for certificate validation.

This feature uses the native urllib3.future implementation. The security properties of a custom resolver depend on the chosen transport and provider. DNSSEC validation is available when supported and enabled by the resolver implementation and provider.

from niquests import Session

with Session(resolver="doh+google://") as s:
    resp = s.get("https://httpbingo.org/get")

from niquests import AsyncSession

async with AsyncSession(resolver="doh+google://") as s:
    resp = await s.get("https://httpbingo.org/get")

from niquests import Session

with Session(resolver=["doh+google://", "doh://cloudflare-dns.com"]) as s:
    resp = s.get("https://httpbingo.org/get")

from niquests import AsyncSession

async with AsyncSession(resolver=["doh+google://", "doh://cloudflare-dns.com"]) as s:
    resp = await s.get("https://httpbingo.org/get")

"doh+google://"  # Shortcut for Google DNS over HTTPS.
"dot+google://"  # Shortcut for Google DNS over TLS.
"doh+cloudflare://"  # Shortcut for Cloudflare DNS over HTTPS.
"doq+adguard://"  # Shortcut for AdGuard DNS over QUIC.
"dou://1.1.1.1"  # DNS over UDP.
"dou://1.1.1.1:8853"  # DNS over UDP on port 8853.
"doh://my-resolver.tld"  # DNS over HTTPS with a custom server.

export NIQUESTS_DNS_URL="doh://google.dns"

from niquests import Session

with Session(resolver="doh+google://default/?verify=false") as s:
    resp = s.get("https://httpbingo.org/get")

from niquests import AsyncSession

async with AsyncSession(resolver="doh+google://default/?verify=false") as s:
    resp = await s.get("https://httpbingo.org/get")

from niquests import Session

with Session(resolver="doh+google://default/?timeout=1") as s:
    resp = s.get("https://httpbingo.org/get")

from niquests import AsyncSession

async with AsyncSession(resolver="doh+google://default/?timeout=1") as s:
    resp = await s.get("https://httpbingo.org/get")

Happy Eyeballs¶

New in version 3.5.5.

The underlying urllib3.future library provides Happy Eyeballs behind one option.

Happy Eyeballs (also called Fast Fallback) is an algorithm published by the IETF that makes dual-stack applications (those that understand both IPv4 and IPv6) more responsive to users by attempting to connect using both IPv4 and IPv6 at the same time (preferring IPv6), thus minimizing common problems experienced by users with imperfect IPv6 connections or setups.

The name “Happy Eyeballs” uses “eyeball” to describe endpoints that represent human Internet users, as opposed to servers.

import niquests

with niquests.Session(happy_eyeballs=True) as s:
    ...

import niquests

async with niquests.AsyncSession(happy_eyeballs=True) as s:
    ...

A single happy_eyeballs=True option enables the algorithm.

OCSP requests for certificate revocation checks also use the configured Happy Eyeballs setting.

WebSockets¶

New in version 3.9: Requires the WebSocket extra: pip install niquests[ws].

WebSockets are a vital part of the web ecosystem alongside HTTP. Niquests provides an integrated interface to reduce the friction of connecting to a WebSocket server for the first time.

from niquests import Session

with Session() as s:
    resp = s.get(
        "wss://echo.websocket.org",
    )

    print(resp.status_code)  # 101 Switching Protocols

    print(resp.extension.next_payload())  # Read the next server message.

    resp.extension.send_payload("Hello World")

    print(resp.extension.next_payload() == "Hello World")  # True

    resp.extension.close()

from niquests import AsyncSession
import asyncio

async def main() -> None:
    async with AsyncSession() as s:
        resp = await s.get("wss://echo.websocket.org")

        # ...

        print(await resp.extension.next_payload())  # unpack the next message from server

        await resp.extension.send_payload("Hello World")

        print((await resp.extension.next_payload()) == "Hello World")  # output True!

        await resp.extension.close()

asyncio.run(main())

from niquests import Session

with Session() as s:
    resp = s.get(
        "wss://echo.websocket.org",
    )

    resp.extension.ping()  # Send a ping to the WebSocket server.

from niquests import AsyncSession

async with AsyncSession() as s:
    resp = await s.get(
        "wss://echo.websocket.org",
    )

    await resp.extension.ping()  # Send a ping to the WebSocket server.

from __future__ import annotations

from niquests import Session, Response, ReadTimeout
from threading import Thread
from time import sleep


def pull_message_from_server(my_response: Response) -> None:
    """Read messages here."""
    iteration_counter = 0

    while my_response.extension.closed is False:
        try:
            # Blocks for at most one second.
            message = my_response.extension.next_payload()

            if message is None:  # server just closed the connection. exit.
                print("received goaway from server")
                return

            print(f"received message: '{message}'")
        except ReadTimeout:  # if no message received within 1s
            pass

        sleep(1)  # let some time for the write part to acquire the lock
        iteration_counter += 1

        # Send a ping every four iterations.
        if iteration_counter % 4 == 0:
            my_response.extension.ping()
            print("ping sent")

if __name__ == "__main__":

    with Session() as s:
        # connect to websocket server "echo.websocket.org" with timeout of 1s (both read and connect)
        resp = s.get("wss://echo.websocket.org", timeout=1)

        if resp.status_code != 101:
            exit(1)

        t = Thread(target=pull_message_from_server, args=(resp,))
        t.start()

        # Send messages here.
        for i in range(30):
            to_send = f"Hello World {i}"
            resp.extension.send_payload(to_send)
            print(f"sent message: '{to_send}'")
            sleep(1)  # let some time for the read part to acquire the lock

        # exit gently!
        resp.extension.close()

        # wait for thread proper exit.
        t.join()

        print("program ended!")

import asyncio
from niquests import AsyncSession, ReadTimeout, Response

async def read_from_ws(my_response: Response) -> None:
    iteration_counter = 0

    while my_response.extension.closed is False:
        try:
            # Blocks for at most one second.
            message = await my_response.extension.next_payload()

            if message is None:  # server just closed the connection. exit.
                print("received goaway from server")
                return

            print(f"received message: '{message}'")
        except ReadTimeout:  # if no message received within 1s
            pass

        await asyncio.sleep(1)  # let some time for the write part to acquire the lock
        iteration_counter += 1

        # Send a ping every four iterations.
        if iteration_counter % 4 == 0:
            await my_response.extension.ping()
            print("ping sent")

async def main() -> None:
    async with AsyncSession() as s:
        resp = await s.get("wss://echo.websocket.org", timeout=1)

        print(resp)

        task = asyncio.create_task(read_from_ws(resp))

        for i in range(30):
            to_send = f"Hello World {i}"
            await resp.extension.send_payload(to_send)
            print(f"sent message: '{to_send}'")
            await asyncio.sleep(1)  # let some time for the read part to acquire the lock

        # exit gently!
        await resp.extension.close()
        await task


if __name__ == "__main__":
    asyncio.run(main())

Server-sent events (SSE)¶

New in version 3.11.2.

Server-Sent Events, commonly abbreviated SSE, provide a standard way to stream events continuously from a server to a client in real time.

from __future__ import annotations

import niquests

if __name__ == "__main__":
    with niquests.Session() as s:
        r = s.post("sse://httpbingo.org/sse")
        print(r.status_code)

        while r.extension.closed is False:
            event: niquests.ServerSentEvent | None = r.extension.next_payload()
            print(event)

import niquests
import asyncio

async def main() -> None:
    async with niquests.AsyncSession() as s:
        r = await s.post("sse://httpbingo.org/sse")

        print(r)  # output: <Response HTTP/2 [200]>

        while r.extension.closed is False:
            print(await r.extension.next_payload())  # ServerSentEvent(event='ping', data='{"id":0,"timestamp":1732857000473}')

if __name__ == "__main__":

    asyncio.run(main())

import niquests

if __name__ == "__main__":
    with niquests.Session() as s:
        r = s.post("sse://sse.dev/test")
        events = []

        while r.extension.closed is False:
            event = r.extension.next_payload()
            if event is None:  # The remote peer closed the stream.
                break

        events.append(event)  # add the event to list

        if len(events) >= 10:  # close ourselves SSE stream & notify remote peer.
            r.extension.close()

import niquests
import asyncio

async def main() -> None:
    async with niquests.AsyncSession() as s:
        r = await s.post("sse://sse.dev/test")

        events = []

        while r.extension.closed is False:
            event = await r.extension.next_payload()

            if event is None:  # The remote peer closed the stream.
                break

            events.append(event)  # add the event to list

            if len(events) >= 10:  # close ourselves SSE stream & notify remote peer.
                await r.extension.close()

if __name__ == "__main__":

    asyncio.run(main())

Unix sockets¶

New in version 3.17.0.

Niquests natively supports connecting to services through Unix domain sockets. This is useful for local services, such as the Docker Engine API, databases, or any application that exposes an HTTP API over a Unix socket.

from niquests import Session

with Session() as s:
    # %2F is the URL-encoded forward slash
    response = s.get("http+unix://%2Fvar%2Frun%2Fdocker.sock/version")
    print(response.json())

from niquests import AsyncSession

async with AsyncSession() as s:
    response = await s.get("http+unix://%2Fvar%2Frun%2Fdocker.sock/version")
    print(response.json())

http+unix://<url-encoded-socket-path>/<api-path>

from concurrent.futures import ThreadPoolExecutor
from niquests import Session

endpoints = ["/containers/json", "/images/json", "/version", "/info"]
with Session() as s, ThreadPoolExecutor() as executor:
    responses = list(executor.map(
        lambda endpoint: s.get(
            f"http+unix://%2Fvar%2Frun%2Fdocker.sock{endpoint}"
        ),
        endpoints,
    ))

import asyncio
from niquests import AsyncSession

async with AsyncSession() as s:
    endpoints = ["/containers/json", "/images/json", "/version", "/info"]
    responses = await asyncio.gather(*(
        s.get(f"http+unix://%2Fvar%2Frun%2Fdocker.sock{endpoint}")
        for endpoint in endpoints
    ))

import niquests

with niquests.Session() as s:
    r = s.get("psse+unix://%2Ftmp%2Fhello.sock/sse")
    while not r.extension.closed:
        print(r.extension.next_payload())

    ws = s.get("ws+unix://%2Ftmp%2Fhello.sock/ws")
    ws.extension.send_payload("Hello")
    print(ws.extension.next_payload())
    ws.extension.close()

import niquests

async with niquests.AsyncSession() as s:
    r = await s.get("psse+unix://%2Ftmp%2Fhello.sock/sse")
    while not r.extension.closed:
        print(await r.extension.next_payload())

    ws = await s.get("ws+unix://%2Ftmp%2Fhello.sock/ws")
    await ws.extension.send_payload("Hello")
    print(await ws.extension.next_payload())
    await ws.extension.close()

WSGI/ASGI application testing¶

New in version 3.17.0.

Niquests provides built-in adapters for testing WSGI and ASGI applications directly without starting a server. This is particularly useful for integration testing.

from fastapi import FastAPI, Request

app = FastAPI()

@app.get("/hello")
async def hello(request: Request):
    return {"message": "hello from asgi"}

@app.api_route("/echo", methods=["GET", "POST"])
async def echo(request: Request):
    body = await request.body()
    return {"body": body.decode()}

import asyncio
from niquests import AsyncSession

async def main():
    async with AsyncSession(app=app) as s:
        resp = await s.get("/hello?foo=bar")
        print(resp.status_code)  # 200
        print(resp.json())  # {"message": "hello from asgi"}

asyncio.run(main())

async def main():
    async with AsyncSession(app=app) as s:
        resp = await s.post("/echo", data=b"foobar", stream=True)

        body = b""
        async for chunk in await resp.iter_content(6):
            body += chunk

        print(body)

asyncio.run(main())

from fastapi import FastAPI, WebSocket
from starlette.responses import StreamingResponse

app = FastAPI()

@app.websocket("/ws-echo")
async def ws_echo(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"echo: {data}")

@app.get("/sse-events")
async def sse_events():
    async def generate():
        for i in range(3):
            yield f"event: message\ndata: event {i}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

from niquests import Session

with Session(app=app) as s:
    resp = s.get("wss://default/ws-echo")

    resp.extension.send_payload("Hello")
    print(resp.extension.next_payload())  # "echo: Hello"

    resp.extension.close()

from niquests import Session

with Session(app=app) as s:
    resp = s.get("sse://default/sse-events")

    while not resp.extension.closed:
        event = resp.extension.next_payload()
        if event is None:
            break
        print(event)  # ServerSentEvent(event='message', data='event 0')

import asyncio
from niquests import AsyncSession

async def main():
    async with AsyncSession(app=app) as s:
        # WebSocket
        resp = await s.get("wss://default/ws-echo")
        await resp.extension.send_payload("Hello")
        print(await resp.extension.next_payload())  # "echo: Hello"
        await resp.extension.close()

        # SSE
        resp = await s.get("sse://default/sse-events")
        while not resp.extension.closed:
            event = await resp.extension.next_payload()
            if event is None:
                break
            print(event)

asyncio.run(main())

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/hello")
def hello():
    return jsonify({"message": "hello from wsgi"})

@app.route("/echo", methods=["GET", "POST"])
def echo():
    return jsonify({"body": request.get_data(as_text=True)})

from niquests import Session

with Session(app=app) as s:
    resp = s.get("/hello?foo=bar")
    print(resp.status_code)  # 200
    print(resp.json())  # {"message": "hello from wsgi"}

with Session(app=app) as s:
    resp = s.post("/echo", data=b"foobar", stream=True)
    print(resp.json())

    for chunk in resp.iter_content(6):
        ...

from flask import Flask, Response

app = Flask(__name__)

@app.route("/sse-events")
def sse_events():
    def generate():
        for i in range(3):
            yield f"event: message\ndata: event {i}\n\n"
    return Response(generate(), mimetype="text/event-stream")

from niquests import Session

with Session(app=app) as s:
    resp = s.get("sse://default/sse-events")

    while not resp.extension.closed:
        event = resp.extension.next_payload()
        if event is None:
            break
        print(event)  # ServerSentEvent(event='message', data='event 0')

Running as a WASI component¶

New in version 3.21.0.

WASI is an excellent fit for sandboxed applications, plug-ins, serverless functions, and edge deployments. A component starts without ambient access to the network or filesystem: its WIT world declares what it can use, and the host decides which of those capabilities to grant when it runs the component. The same component can therefore be deployed under different security policies without changing its Python code.

For the most complete Niquests experience, prefer the WASI socket interfaces. Use Preview 2 sockets for synchronous code and Preview 3 sockets for asynchronous code. The matching wasi:cli/command world from the componentize-py source distribution includes those interfaces.

Install Niquests with the Rustls backend so that HTTPS can run over WASI sockets, then download the matching componentize-py source archive:

$ python -m pip install "niquests[rtls]" "componentize-py==0.25.0"
$ python -m pip download --no-deps --no-binary=:all: "componentize-py==0.25.0"
$ python -m tarfile -e componentize_py-0.25.0.tar.gz .

Choose the execution model for your component:

import niquests
from wit_world import exports


class Run(exports.Run):
    def run(self) -> None:
        response = niquests.get("https://httpbingo.org/get", timeout=10)
        print(response.status_code)

$ componentize-py -d componentize_py-0.25.0/wit -w wasi:cli/command@0.2.0 componentize app -o app.wasm
$ wasmtime run -Sinherit-network -Sallow-ip-name-lookup=y app.wasm

import niquests
from wit_world import exports


class Run(exports.Run):
    async def run(self) -> None:
        response = await niquests.aget("https://httpbingo.org/get", timeout=10)
        print(response.status_code)

$ componentize-py -d componentize_py-0.25.0/wit -w wasi:cli/command@0.3.0 componentize app -o app.wasm
$ wasmtime run -Sp3 -Sinherit-network -Sallow-ip-name-lookup=y app.wasm

The example grants network access and permission to resolve names, but no host directory is preopened. Do not add --dir unless the application genuinely needs host files. Be aware that Wasmtime’s inherit-network grant is intentionally broad: it exposes the host network namespace rather than only the URL shown above. In production, prefer a host-level destination allowlist over unrestricted inherited network access when your runtime supports one.

Niquests selects the available WASI transport automatically; application code does not mount an adapter. Socket WIT provides native connection pooling, protocol negotiation, WebSocket, SSE, redirects, and the usual Session behavior. A host-managed WASI HTTP interface is also supported as a constrained fallback. See WASI transports and capabilities before choosing that contract or designing a least-authority deployment.

Running in the browser (Pyodide)¶

New in version 3.18.0.

Niquests runs natively in Pyodide without configuration changes. HTTP requests, WebSocket, and SSE use Session, AsyncSession, and resp.extension. The adapter is selected automatically when Pyodide is detected.

# This exact code works in both CPython and Pyodide:
import niquests

resp = niquests.get("https://httpbingo.org/get")
print(resp.json())

# This exact code works in both CPython and Pyodide:
import niquests

resp = await niquests.aget("https://httpbingo.org/get")
print(resp.json())

WebSocket and SSE use the same API as described in the sections above:

from niquests import Session

with Session() as s:
    # Behind the scenes, uses the browser's native WebSocket API
    resp = s.get("wss://echo.websocket.org")
    resp.extension.send_payload("Hello")
    print(resp.extension.next_payload())
    resp.extension.close()

    # Behind the scenes, uses fetch streaming under the hood
    resp = s.get("sse://some-server.example/events")
    while not resp.extension.closed:
        event = resp.extension.next_payload()
        if event is None:
            break
        print(event)

from niquests import AsyncSession

async with AsyncSession() as s:
    # Behind the scenes, uses the browser's native WebSocket API
    resp = await s.get("wss://echo.websocket.org")
    await resp.extension.send_payload("Hello")
    print(await resp.extension.next_payload())
    await resp.extension.close()

    # Behind the scenes, uses fetch streaming under the hood
    resp = await s.get("sse://some-server.example/events")
    while not resp.extension.closed:
        event = await resp.extension.next_payload()
        if event is None:
            break
        print(event)