Niquests Extension Ecosystem¶
Native Niquests Extensions¶
The community regularly produce useful 3rd party extension that plugs directly into Niquests APIs without patch. We do not maintain the showcased extensions, any issue encountered with listed extensions must be reported to the extension owner.
niquests-cache¶
niquests-cache is a companion library that adds HTTP response caching to Niquests with
pluggable storage backends. It is intended as the successor to requests-cache for projects that
have migrated to Niquests, but unlike requests-cache it supports async sessions natively. It will
also eliminate the extra requests dependency brought in by requests-cache.
Three backends are included:
SQLite (default) — efficient structured storage via a single
.sqlitefileFilesystem — one file per cached entry, human-inspectable
In-memory — dictionary-backed, no disk I/O
Install it from PyPI:
pip install niquests-cache
For synchronous use, the API is the same and async is almost identical:
import niquests_cache
session = niquests_cache.CachedSession('demo_cache')
session.get('https://httpbin.org/delay/1')
import asyncio
import niquests_cache
async def main() -> None:
async with niquests_cache.AsyncCachedSession('demo_cache') as session:
await session.get('https://httpbin.org/delay/1')
asyncio.run(main())
CachedSession extends niquests.Session and
AsyncCachedSession extends niquests.AsyncSession.
The cached_session helper returns a CachedSession or
AsyncCachedSession depending on the aio parameter. Primarily its use is its app_name
parameter that makes it easy to store cache in conventional platform-specific paths without having to specify or calculate
the complete path.
from niquests_cache import cached_session
with cached_session(app_name='my-tool') as session: # On XDG, stores to ~/.cache/my-tool/http.sqlite
response = session.get('https://httpbin.org/get')
response.raise_for_status()
import asyncio
from niquests_cache import cached_session
async def main() -> None:
async with cached_session(app_name='my-tool', aio=True) as session:
response = await session.get('https://httpbin.org/get')
response.raise_for_status()
asyncio.run(main())
You can select a backend by alias or by passing an instance directly:
from niquests_cache import CachedSession
session = CachedSession('my_cache', backend='filesystem')
session = CachedSession('my_cache', backend='memory')
session = CachedSession('my_cache', backend='sqlite') # default
Individual requests accept per-request cache controls:
session.get('https://example.com', force_refresh=True) # bypass cache, store fresh response
session.get('https://example.com', only_if_cached=True) # return 504 if not cached
session.get('https://example.com', expire_after=60) # override session TTL
Cache behaviour is tuneable at runtime through session.settings, and caching can be temporarily
suspended with a context manager:
with session.cache_disabled():
response = session.get('https://example.com') # not cached
For the full configuration reference (URL-pattern TTL overrides, custom cache keys, header matching, ETag/Last-Modified revalidation, and more), see the niquests-cache documentation.
niquests-mock¶
niquests-mock is a RESPX-style HTTP mocking library built specifically for Niquests. It provides pytest fixtures, decorators, and context managers for mocking Niquests calls without importing Requests or patching the Requests ecosystem.
Install it from PyPI:
pip install niquests-mock
The pytest plugin exposes a niquests_mock fixture:
import niquests
def test_fixture_style(niquests_mock):
route = niquests_mock.get("https://example.org/")
route.respond(status_code=200)
response = niquests.get("https://example.org/")
assert route.called
assert response.status_code == 200
It can also be used as a decorator:
import niquests
import niquests_mock as nmock
@nmock.mock
def test_decorator_style():
route = nmock.get("https://example.org/", name="homepage").respond(status_code=200)
response = niquests.get("https://example.org/")
route.assert_called_once()
assert nmock.lookup("homepage") is route
assert response.status_code == 200
Extensions from Requests to Niquests¶
One of the main strength behind Requests is the wide range of community plugins / extensions that allows one to extend its abilities.
In this chapter, we’ll look into some of the most populars extensions and see how you can use it with Niquests instead of Requests.
There is 4 levels of compatibility:
Native: No adjustment are required, should work as-is out of the box.
Working: A minor tweak is required that does not harm your environment at all.
Usable: A consequent patch may be required, usually assigning / injecting Niquests module impersonating Requests.
Unusable: You cannot use the plugin due to various reasons. Usually the library leverage private properties or excessively rely on broken behaviors that are now patched in Niquests.
Warning
Sometimes, plugin/packages explicitly require Requests as a dependency, thus making your environment heavier, sometime for no reason. No solution exist to override this behavior.
Note
Feel free to reach out to the maintainers and speak up about Niquests. Suggesting a patch to support both Requests, and Niquests is really straightforward!
Requests Cache¶
Note
Classified as: Working
requests-cache is a persistent HTTP cache that provides an easy way to get better performance with the python requests library.
Quickstart to leverage its potential:
import requests_cache
import niquests
class CacheSession(requests_cache.session.CacheMixin, niquests.Session):
...
if __name__ == "__main__":
s = CacheSession()
for i in range(60):
r = s.get('https://httpbin.org/delay/1')
Warning
Be advised that this extension nullify the advantage of using multiplexed=True within your Session constructor as is eagerly access the content.
Tip
niquests-cache is a native alternative.
responses¶
Note
Classified as: Usable
Apply the following code to your conftest.py:
from sys import modules
import requests
import niquests
from niquests.packages import urllib3
# responses is tied to Requests
# and Niquests is entirely compatible with it.
# we can fool it without effort.
modules["requests"] = niquests
modules["requests.adapters"] = niquests.adapters
modules["requests.models"] = niquests.models
modules["requests.exceptions"] = niquests.exceptions
modules["requests.packages.urllib3"] = urllib3
If you want to extend Responses to support mocking asynchronous calls, you will have to extend the bellow code with the following:
# make 'responses' mock both sync and async
# 'Requests' ever only supported sync
# Fortunately interfaces are mirrored in 'Niquests'
from unittest import mock as std_mock # noqa: E402
import responses # noqa: E402
class NiquestsMock(responses.RequestsMock):
"""Asynchronous support for responses"""
def __init__(self, *args, **kwargs) -> None:
super().__init__(
*args,
target="niquests.adapters.HTTPAdapter.send",
**kwargs,
)
self._patcher_async = None
def unbound_on_async_send(self):
async def send(
adapter: "niquests.adapters.AsyncHTTPAdapter",
request: "niquests.PreparedRequest",
*args: typing.Any,
**kwargs: typing.Any,
) -> "niquests.Response":
if args:
# that probably means that the request was sent from the custom adapter
# It is fully legit to send positional args from adapter, although,
# `requests` implementation does it always with kwargs
# See for more info: https://github.com/getsentry/responses/issues/642
try:
kwargs["stream"] = args[0]
kwargs["timeout"] = args[1]
kwargs["verify"] = args[2]
kwargs["cert"] = args[3]
kwargs["proxies"] = args[4]
except IndexError:
# not all kwargs are required
pass
resp = self._on_request(adapter, request, **kwargs)
if kwargs["stream"]:
return resp
resp.__class__ = niquests.Response
return resp
return send
def unbound_on_send(self):
def send(
adapter: "niquests.adapters.HTTPAdapter",
request: "niquests.PreparedRequest",
*args: typing.Any,
**kwargs: typing.Any,
) -> "niquests.Response":
if args:
# that probably means that the request was sent from the custom adapter
# It is fully legit to send positional args from adapter, although,
# `requests` implementation does it always with kwargs
# See for more info: https://github.com/getsentry/responses/issues/642
try:
kwargs["stream"] = args[0]
kwargs["timeout"] = args[1]
kwargs["verify"] = args[2]
kwargs["cert"] = args[3]
kwargs["proxies"] = args[4]
except IndexError:
# not all kwargs are required
pass
return self._on_request(adapter, request, **kwargs)
return send
def start(self) -> None:
if self._patcher:
# we must not override value of the _patcher if already applied
# this prevents issues when one decorated function is called from
# another decorated function
return
self._patcher = std_mock.patch(target=self.target, new=self.unbound_on_send())
self._patcher_async = std_mock.patch(
target=self.target.replace("HTTPAdapter", "AsyncHTTPAdapter"),
new=self.unbound_on_async_send()
)
self._patcher.start()
self._patcher_async.start()
def stop(self, allow_assert: bool = True) -> None:
if self._patcher:
# prevent stopping unstarted patchers
self._patcher.stop()
self._patcher_async.stop()
# once patcher is stopped, clean it. This is required to create a new
# fresh patcher on self.start()
self._patcher = None
self._patcher_async = None
if not self.assert_all_requests_are_fired:
return
if not allow_assert:
return
not_called = [m for m in self.registered() if m.call_count == 0]
if not_called:
raise AssertionError(
"Not all requests have been executed {!r}".format(
[(match.method, match.url) for match in not_called]
)
)
mock = _default_mock = NiquestsMock(assert_all_requests_are_fired=False)
setattr(responses, "mock", mock)
setattr(responses, "_default_mock", _default_mock)
for kw in [
"activate",
"add",
"_add_from_file",
"add_callback",
"add_passthru",
"assert_call_count",
"calls",
"delete",
"DELETE",
"get",
"GET",
"head",
"HEAD",
"options",
"OPTIONS",
"patch",
"PATCH",
"post",
"POST",
"put",
"PUT",
"registered",
"remove",
"replace",
"reset",
"response_callback",
"start",
"stop",
"upsert",
]:
if not hasattr(responses, kw):
continue
setattr(responses, kw, getattr(mock, kw))
This will automatically make Responses work seamlessly when using awaitable http calls.
betamax¶
Note
Classified as: Usable
Apply the following code to your conftest.py:
from sys import modules
import requests
import niquests
import niquests.packages import urllib3
# betamax is tied to Requests
# and Niquests is almost entirely compatible with it.
# we can fool it without effort.
modules["requests"] = niquests
modules["requests.adapters"] = niquests.adapters
modules["requests.models"] = niquests.models
modules["requests.exceptions"] = niquests.exceptions
modules["requests.packages.urllib3"] = urllib3
# niquests no longer have a compat submodule
# but betamax need it. no worries, as betamax
# explicitly need requests, we'll give it to him.
modules["requests.compat"] = requests.compat
# doing the import now will make betamax working with Niquests!
# no extra effort.
import betamax
# the base mock does not implement close(), which is required
# for our HTTP client. No biggy.
betamax.mock_response.MockHTTPResponse.close = lambda _: None
And make sure that the betamax plugin isn’t loaded at boot with (pyproject.toml):
[tool.pytest.ini_options]
# this avoids pytest loading betamax+Requests at boot.
# this allows us to patch betamax and makes it use Niquests instead.
addopts = "-p no:pytest-betamax"
Or run pytest directly with pytest -p no:pytest-betamax.
Requests-Toolbelt¶
Note
Classified as: Usable
Requests-Toolbelt is a collection of utilities that some users of Niquests may desire, but do not belong in Niquests proper. This library is actively maintained by members of the Requests core team, and reflects the functionality most requested by users within the community.
requests-aws4auth¶
Note
Classified as: Native
requests-file¶
Note
Classified as: Usable
requests-mock¶
Note
Classified as: Usable
You will need to create a fixture to override the default bind to Requests in conftest.py like so:
from sys import modules
import requests
import niquests
from niquests.packages import urllib3
# impersonate Requests!
modules["requests"] = niquests
modules["requests.adapters"] = niquests.adapters
modules["requests.models"] = niquests.models
modules["requests.exceptions"] = niquests.exceptions
modules["requests.packages.urllib3"] = urllib3
modules["requests.compat"] = requests.compat
@pytest.fixture(scope='function')
def patched_requests_mock():
"""This is required because pytest load plugins at boot, way before conftest.
The only reliable way to make requests_mock use Niquests is to customize it after."""
import requests_mock # noqa: E402
class _WrappedMocker(requests_mock.Mocker):
"""Ensure requests_mock work with the drop-in replacement Niquests!"""
def __init__(self, session=None, **kwargs):
# we purposely skip invoking super() to avoid the strict typecheck on session.
self._mock_target = session or niquests.Session
self.case_sensitive = kwargs.pop('case_sensitive', self.case_sensitive)
self._adapter = (
kwargs.pop('adapter', None)
or requests_mock.adapter.Adapter(case_sensitive=self.case_sensitive)
)
self._json_encoder = kwargs.pop('json_encoder', None)
self.real_http = kwargs.pop('real_http', False)
self._last_send = None
if kwargs:
raise TypeError('Unexpected Arguments: %s' % ', '.join(kwargs))
def request(self, *args, **kwargs):
if "response_list" not in kwargs:
if "headers" not in kwargs:
kwargs["headers"] = {}
if "json" in kwargs and kwargs["json"] is not None:
kwargs["headers"]["Content-Type"] = "application/json"
else:
for resp in kwargs["response_list"]:
if "headers" not in resp:
resp["headers"] = {}
if "json" in resp and resp["json"] is not None:
resp["headers"]["Content-Type"] = "application/json"
return self.register_uri(*args, **kwargs)
with _WrappedMocker() as m:
yield m
Then, use it as you were used to:
def test_sometime(patched_requests_mock):
patched_requests_mock.get("https://example.com/", text="hello world")
Warning
This extension load/import Requests at pytest startup.
Disable the plugin auto-loading first by either passing PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 (in environment)
or pytest -p "no:requests_mock" in CLI parameters. You may also append -p "no:requests_mock" in addopts
of your pyproject.toml or equivalent.
requests-ntlm¶
Note
Classified as: Native
requests-unixsocket¶
Note
Classified as: Usable
Warning
Since Niquests 3.17.0 you are able to connect to unixsocket without it, natively.
requests-futures¶
Warning
Classified as: Unusable
This project is no longer required for you! Niquests ships with native asyncio support. Furthermore, you may leverage multiplexing to optimize your HTTP calls at will.
requests-kerberos¶
Note
Classified as: Native
Nothing change from your previous code:
>>> import niquests
>>> from requests_kerberos import HTTPKerberosAuth
>>> r = niquests.get("http://example.org", auth=HTTPKerberosAuth())
The HTTPKerberosAuth can be used natively without patch.
requests-pkcs12¶
Note
Classified as: Native
requests-ntlm3¶
Note
Classified as: Native
requests-gssapi¶
Note
Classified as: Native
Requests-OAuthlib¶
Note
Classified as: Working
requests-oauthlib makes it possible to do the OAuth dance from Niquests automatically. This is useful for the large number of websites that use OAuth to provide authentication. It also provides a lot of tweaks that handle ways that specific OAuth providers differ from the standard specifications.
Please patch your program as follow:
import niquests
from oauthlib.oauth2 import BackendApplicationClient
import requests_oauthlib
requests_oauthlib.OAuth2Session.__bases__ = (niquests.Session,)
client_id = "xxxxxxxxxxxxxxxxxxxxxxxxxxx"
client_secret = "xxxxxxxxxxxxxxxxxxxxxxxxxxx"
token_url = 'https://api.github.com/token'
if __name__ == "__main__":
client = BackendApplicationClient(client_id=client_id)
sample = requests_oauthlib.OAuth2Session(client=client)
token = sample.fetch_token(token_url, client_secret=client_secret)
The key element to be considered is requests_oauthlib.OAuth2Session.__bases__ = (niquests.Session,).
You may apply it to requests_oauthlib.OAuth1Session too.
vcrpy¶
Note
Classified as: Native
vcrpy records and replays HTTP interactions as cassettes, making tests deterministic and fast. It works with Niquests out of the box because it hooks into urllib3 at the connection-pool level.
Basic synchronous usage requires no special setup:
import niquests
import vcr
with vcr.VCR().use_cassette("cassette.yaml"):
r = niquests.get("https://example.com/path")
Enabling async support¶
vcrpy does not patch AsyncHTTPConnectionPool shipped by urllib3.future.
The fixture below extends every cassette context so that async calls are
also played back and recorded.
Add the following to your conftest.py:
import functools
import itertools
import logging
from unittest import mock
import pytest
from urllib3 import AsyncHTTPConnectionPool, AsyncHTTPResponse, HTTPHeaderDict
from vcr.errors import CannotOverwriteExistingCassetteException
from vcr.patch import CassettePatcherBuilder
from vcr.request import Request
log = logging.getLogger(__name__)
_original_async_urlopen = AsyncHTTPConnectionPool.urlopen
_original_build = CassettePatcherBuilder.build
_DEFAULT_PORTS = {"https": 443, "http": 80}
def _serialize_headers(headers) -> dict[str, list[str]]:
out: dict[str, list[str]] = {}
for key, val in headers.items():
out.setdefault(key, []).append(val)
return out
def _deserialize_headers(headers: dict[str, list[str]]) -> HTTPHeaderDict:
hd = HTTPHeaderDict()
for key, values in headers.items():
if isinstance(values, list):
for v in values:
hd.add(key, v)
else:
hd.add(key, values)
return hd
def _reconstruct_uri(pool, url: str) -> str:
if url.startswith(("http://", "https://")):
return url
scheme = pool.scheme or "http"
host = pool.host
port = pool.port
port_postfix = f":{port}" if port and port != _DEFAULT_PORTS.get(scheme) else ""
return f"{scheme}://{host}{port_postfix}{url}"
def _build_async_response(vcr_response) -> AsyncHTTPResponse:
headers = dict(vcr_response.get("headers", {}))
for k in [h for h in headers if h.lower() == "transfer-encoding"]:
del headers[k]
body = vcr_response.get("body", {}).get("string", b"")
if isinstance(body, str):
body = body.encode("utf-8")
return AsyncHTTPResponse(
body=body,
status=vcr_response["status"]["code"],
reason=vcr_response["status"]["message"],
headers=_deserialize_headers(headers),
preload_content=True,
)
def _make_async_urlopen(cassette):
@functools.wraps(_original_async_urlopen)
async def vcr_async_urlopen(self, method, url, body=None, headers=None, **kw):
if headers is None:
headers = {}
uri = _reconstruct_uri(self, url)
vcr_request = Request(method, uri, body, _serialize_headers(headers))
if cassette.can_play_response_for(vcr_request):
log.info("Playing response for %s from cassette", vcr_request)
return _build_async_response(cassette.play_response(vcr_request))
if cassette.write_protected and cassette.filter_request(vcr_request):
raise CannotOverwriteExistingCassetteException(
cassette=cassette, failed_request=vcr_request,
)
log.info("%s not in cassette, sending to real server", vcr_request)
response = await _original_async_urlopen(
self, method, url, body=body, headers=headers, **kw
)
resp_body = response._body
if resp_body is None:
resp_body = await response.data
if isinstance(resp_body, str):
resp_body = resp_body.encode("utf-8")
cassette.append(vcr_request, {
"status": {"code": response.status, "message": response.reason or ""},
"headers": _serialize_headers(response.headers),
"body": {"string": resp_body or b""},
})
return response
return vcr_async_urlopen
def _patched_build(self):
return itertools.chain(
_original_build(self),
(mock.patch.object(
AsyncHTTPConnectionPool, "urlopen",
_make_async_urlopen(self._cassette),
),),
)
@pytest.fixture(scope="session", autouse=True)
def _vcr_async_urllib3():
"""Monkey-patch vcrpy so cassettes also cover async urllib3."""
CassettePatcherBuilder.build = _patched_build
yield
CassettePatcherBuilder.build = _original_build
Then use it in your async tests as you would normally:
import vcr
@vcr.use_cassette("cassette.yaml")
async def test_async_request():
async with niquests.AsyncSession() as s:
r = await s.get("https://example.com/path")
assert r.status_code == 200