Niquests

https://niquests.readthedocs.io/en/latest/
Admin
Niquests is an elegant and simple HTTP library for Python designed as a drop-in replacement for...
Tokens:249,514
Snippets:1,881
Trust Score:8.3
Update:5 months ago
Show doc for...
Tokens:
Raw
### Quickstart WebSocket Connection (Async)

Source: https://niquests.readthedocs.io/en/latest/user/quickstart.html

Establish an asynchronous WebSocket connection to an echo server and interact with messages. Requires `pip install niquests[ws]`. Available since version 3.9+.

```python
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")  # automatically sends a text message to the server

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

        await resp.extension.close()
```

--------------------------------

### Quickstart WebSocket Connection (Sync)

Source: https://niquests.readthedocs.io/en/latest/user/quickstart.html

Establish a synchronous WebSocket connection to an echo server and interact with messages. Requires `pip install niquests[ws]`. Available since version 3.9+.

```python
from niquests import Session

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

    print(resp.status_code)  # it says "101", for "Switching Protocol"

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

    resp.extension.send_payload("Hello World")  # automatically sends a text message to the server

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

    resp.extension.close()  # don't forget this call to release the connection!
```

--------------------------------

### HTTPAdapter usage example

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/adapters.html

Example of creating an HTTPAdapter with retries and mounting it to a session for HTTP URLs.

```python
import niquests
>>> s = niquests.Session()
>>> a = niquests.adapters.HTTPAdapter(max_retries=3)
>>> s.mount('http://', a)
```

--------------------------------

### Install Niquests with HTTP/3 support

Source: https://niquests.readthedocs.io/en/latest/user/install.html

Install Niquests with the 'http3' extra to ensure the 'qh3' dependency is installed for HTTP/3 support. The 'ocsp' extra is an alias for this.

```bash
$ python -m pip install niquests[http3]

```

--------------------------------

### Install Niquests with urllib3-future (PDM Option 1)

Source: https://niquests.readthedocs.io/en/latest/community/faq.html

Install Niquests using PDM with environment variables to ensure urllib3-future is not installed as a binary.

```bash
$ URLLIB3_NO_OVERRIDE=1 PDM_NO_BINARY=urllib3-future pdm add niquests
```

--------------------------------

### Install Niquests with all optional features

Source: https://niquests.readthedocs.io/en/latest/user/install.html

Install Niquests with the 'full' extra to include all available optional features and dependencies.

```bash
$ python -m pip install niquests[full]

```

--------------------------------

### Install niquests from Source

Source: https://niquests.readthedocs.io/en/latest/user/install.html

After obtaining the source code, navigate to the niquests directory and use pip to install the package locally.

```bash
cd niquests
python -m pip install .
```

--------------------------------

### Install Niquests with WebSocket support

Source: https://niquests.readthedocs.io/en/latest/user/install.html

Install Niquests with the 'ws' extra to enable integrated WebSocket functionality.

```bash
$ python -m pip install niquests[ws]

```

--------------------------------

### Install Niquests with urllib3-future (Poetry Option 2)

Source: https://niquests.readthedocs.io/en/latest/community/faq.html

Install Niquests using Poetry with specific environment variables to prevent urllib3-future from being installed as a binary.

```bash
$ URLLIB3_NO_OVERRIDE=1 POETRY_INSTALLER_NO_BINARY=urllib3-future poetry add niquests
```

--------------------------------

### Install Niquests

Source: https://niquests.readthedocs.io/en/latest/user/install.html

Install the Niquests package using pip. This command installs the core package and its mandatory dependencies.

```bash
$ python -m pip install niquests

```

--------------------------------

### Install Niquests with SOCKS proxy support

Source: https://niquests.readthedocs.io/en/latest/user/install.html

Install Niquests with the 'socks' extra to enable SOCKS proxy support.

```bash
$ python -m pip install niquests[socks]

```

--------------------------------

### Install SOCKS Proxy Dependencies

Source: https://niquests.readthedocs.io/en/latest/user/advanced.html

Install the necessary third-party libraries for SOCKS proxy support using pip.

```bash
$ python -m pip install niquests[socks]
```

--------------------------------

### Making a GET Request

Source: https://niquests.readthedocs.io/en/latest/user/advanced.html

Demonstrates how to retrieve data from a URL using the GET method and inspect the response.

```APIDOC
## GET

### Description
Retrieves a resource from a given URL. This is an idempotent method used for fetching data.

### Method
GET

### Endpoint
`https://api.github.com/repos/psf/requests/git/commits/a050faf084662f3a352dd1a941f2c7c9f886d4ad`

### Request Example
```python
import niquests
r = niquests.get('https://api.github.com/repos/psf/requests/git/commits/a050faf084662f3a352dd1a941f2c7c9f886d4ad')
```

### Response
#### Success Response (200)
- **content-type** (string) - The type of content returned, e.g., `application/json; charset=utf-8`.
- **json** (object) - Parsed JSON response body.

#### Response Example
```python
if r.status_code == niquests.codes.ok:
    print(r.headers['content-type'])
    commit_data = r.json()
    print(commit_data.keys())
    print(commit_data['committer'])
    print(commit_data['message'])
```
```

--------------------------------

### Install Niquests with Rustls backend

Source: https://niquests.readthedocs.io/en/latest/user/install.html

Install Niquests with the 'rtls' extra to swap the default OpenSSL/LibreSSL backend for a Rustls backend.

```bash
$ python -m pip install niquests[rtls]

```

--------------------------------

### Install niquests-cache

Source: https://niquests.readthedocs.io/en/latest/community/extensions.html

Install the niquests-cache library using pip. This library adds HTTP response caching to Niquests.

```bash
pip install niquests-cache

```

--------------------------------

### Install Requests with chardet for Python 3

Source: https://niquests.readthedocs.io/en/latest/community/updates.html

Use this command to install Requests with the `use_chardet_on_py3` extra, ensuring compatibility with chardet for Python 3.

```bash
pip install "requests[use_chardet_on_py3]"
```

--------------------------------

### Make a GET Request

Source: https://niquests.readthedocs.io/en/latest/user/advanced.html

This snippet shows how to make a simple GET request to a URL and store the response.

```python
>>> r = niquests.get('https://en.wikipedia.org/wiki/Monty_Python')
```

--------------------------------

### Install Niquests with speedup extras

Source: https://niquests.readthedocs.io/en/latest/user/install.html

Install Niquests with the 'speedups' extra to enable faster decompression with Brotli and Zstandard, and use the 'orjson' decoder.

```bash
$ python -m pip install niquests[speedups]

```

--------------------------------

### Install Niquests with specific speedup (Zstandard)

Source: https://niquests.readthedocs.io/en/latest/user/install.html

Install Niquests with only the 'zstd' extra, for Zstandard decompression, excluding other speedup options like 'orjson'.

```bash
$ python -m pip install niquests[zstd]

```

--------------------------------

### Create Image from Binary Content

Source: https://niquests.readthedocs.io/en/latest/user/quickstart.html

Example demonstrating how to create an image object from binary response content using Pillow and BytesIO.

```python
>>> from PIL import Image
>>> from io import BytesIO

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

--------------------------------

### AsyncSession Basic Usage

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/async_session.html

Demonstrates the basic usage of AsyncSession to make an asynchronous GET request. Requires importing the niquests library.

```python
import niquests

s = niquests.AsyncSession()
await s.get('https://httpbin.org/get')
```

--------------------------------

### Initializing AsyncSession with Adapters

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/async_session.html

Demonstrates how to initialize an AsyncSession and mount different adapters for various URL schemes, including HTTP, HTTPS, SSE, PSSE, WS, and WSS. It also shows mounting an adapter for ASGI applications.

```python
wasm_adapter = AsyncPyodideAdapter(
    max_retries=retries,
)
for supported_scheme in ["http", "https", "sse", "psse", "ws", "wss"]:
    self.mount(f"{supported_scheme}://", wasm_adapter)
if app is not None:
    adapter = AsyncServerGatewayInterface(
        app=app,
        max_retries=retries,
    )
    self.mount("asgi://default", adapter)
    for _scheme in ("ws", "wss", "sse", "psse"):
        self.mount(f"{_scheme}://default", adapter)
    if self.base_url is None:
        self.base_url = "asgi://default"
```

--------------------------------

### AsyncHTTPDigestAuth Example Usage

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/auth.html

Example of how to use AsyncHTTPDigestAuth with an AsyncSession to perform a GET request to a digest-authenticated endpoint.

```python
>>> import niquests
>>> auth = niquests.auth.AsyncHTTPDigestAuth('user', 'pass')
>>> async with niquests.AsyncSession() as session:
...     r = await session.get('https://httpbin.org/digest-auth/auth/user/pass', auth=auth)
...     print(r.status_code)
200
```

--------------------------------

### Async Function Boilerplate

Source: https://niquests.readthedocs.io/en/latest/user/quickstart.html

All async examples must be enclosed in a proper async function and started by asyncio.run(...).

```python
import asyncio
import niquests

async def main() -> None:
    """paste your example code here!"""

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

```

--------------------------------

### Create HTTPX Client Instance

Source: https://niquests.readthedocs.io/en/latest/dev/httpx.html

This shows the standard way to instantiate a client in HTTPX.

```python
client = httpx.Client(**kwargs)
```

--------------------------------

### Rebuild Request Method Based on Redirects

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/sessions.html

Adjusts the request method when being redirected, following RFC specifications and common browser behaviors. For example, it changes 'SEE OTHER' and 'FOUND' redirects to 'GET', and 'POST' redirects with 'MOVED' status to 'GET'.

```python
def rebuild_method(self, prepared_request: PreparedRequest, response: Response):
    """When being redirected we may want to change the method of the request
    based on certain specs or browser behavior.
    """
    method = prepared_request.method

    # https://tools.ietf.org/html/rfc7231#section-6.4.4
    if response.status_code == codes.see_other and method != "HEAD":  # type: ignore[attr-defined]
        method = "GET"

    # Do what the browsers do, despite standards...
    # First, turn 302s into GETs.
    if response.status_code == codes.found and method != "HEAD":  # type: ignore[attr-defined]
        method = "GET"

    # Second, if a POST is responded to with a 301, turn it into a GET.
    # This bizarre behaviour is explained in Issue 1704.
    if response.status_code == codes.moved and method == "POST":  # type: ignore[attr-defined]
        method = "GET"

    prepared_request.method = method
```

--------------------------------

### Mounting AsyncHTTPAdapter to a Session

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/adapters.html

Demonstrates how to create an AsyncSession, initialize an AsyncHTTPAdapter with a specific number of retries, and mount the adapter to handle requests for a given URL scheme.

```python
>>> import niquests
>>> s = niquests.AsyncSession()
>>> a = niquests.adapters.AsyncHTTPAdapter(max_retries=3)
>>> s.mount('http://', a)
```

--------------------------------

### Basic Async Request with AsyncSession

Source: https://niquests.readthedocs.io/en/latest/user/quickstart.html

A fundamental example of using AsyncSession to perform asynchronous HTTP requests within an asyncio event loop. Note that top-level async functions like get() are not supported.

```python
import asyncio
from niquests import AsyncSession, Response

async def fetch(url: str) -> Response:
    async with AsyncSession() as s:
        return await s.get(url)

async def main() -> None:
    tasks = []

    for _ in range(10):
        tasks.append(asyncio.create_task(fetch("https://httpbingo.org/delay/1")))

    responses = await asyncio.gather(*tasks)

    print(responses)

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

```

--------------------------------

### AsyncSession Initialization and Adapter Mounting

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/async_session.html

Demonstrates how AsyncSession is initialized with various adapters for different URL schemes like HTTP, HTTPS, SSE, PSSE, WS, and WSS. It also shows how to mount ASGI applications.

```APIDOC
## AsyncSession Initialization

### Description
Initializes an AsyncSession with optional ASGI application, retry configurations, and mounts adapters for various schemes.

### Method
`__init__`

### Parameters
- `app` (optional): An ASGI application to mount.
- `retries` (optional): The number of retries for requests.

### Example
```python
from niquests import AsyncSession

# Initialize with default settings
session = AsyncSession()

# Initialize with custom retries
session_with_retries = AsyncSession(retries=3)

# Initialize with an ASGI app
async def my_asgi_app(scope, receive, send):
    pass
session_with_asgi = AsyncSession(app=my_asgi_app)
```

## Adapter Mounting

### Description
AsyncSession automatically mounts adapters for different URL schemes. It supports `http`, `https`, `sse`, `psse`, `ws`, `wss` via `AsyncPyodideAdapter` or `AsyncServerGatewayInterface` for ASGI applications.

### Method
`mount(prefix: str, adapter: AsyncBaseAdapter)`

### Parameters
- `prefix` (str): The URL scheme prefix to mount the adapter to (e.g., "http://").
- `adapter` (AsyncBaseAdapter): The adapter instance to mount.

### Example
```python
from niquests import AsyncSession, AsyncHTTPAdapter

session = AsyncSession()
http_adapter = AsyncHTTPAdapter()
session.mount("http://", http_adapter)
```
```

--------------------------------

### Install Niquests with urllib3-future (pip)

Source: https://niquests.readthedocs.io/en/latest/community/faq.html

Use this command to install Niquests while preventing urllib3-future from being installed as a binary, ensuring proper cohabitation.

```bash
$ URLLIB3_NO_OVERRIDE=1 pip install niquests --no-binary urllib3-future
```

--------------------------------

### Usage Example of PreparedRequest

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/models.html

Illustrates how to create a Request object, prepare it into a PreparedRequest, and then send it using a Session. This shows the typical workflow for making an HTTP request.

```python
>>> import niquests
>>> req = niquests.Request('GET', 'https://httpbin.org/get')
>>> r = req.prepare()
>>> r
<PreparedRequest [GET]>

>>> s = niquests.Session()
>>> s.send(r)
<Response HTTP/2 [200]>
```

--------------------------------

### Create and Prepare a Request

Source: https://niquests.readthedocs.io/en/latest/api.html

Instantiate a Request object with a method and URL, then prepare it for transmission.

```python
>>> import niquests
>>> req = niquests.Request('GET', 'https://httpbin.org/get')
>>> req.prepare()
<PreparedRequest [GET]>
```

--------------------------------

### Session Initialization with Adapters

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/sessions.html

Demonstrates how a Session object is initialized with various adapters for different protocols (HTTP, WASM, ASGI, WSGI). This setup determines how requests are handled based on the URL scheme.

```python
if wasm_adapter:
            self.mount("wasm://", wasm_adapter)
        else:
            self.mount("http://", Adapter(max_retries=retries, resolver=resolver, source_address=source_address, disable_http1=disable_http1, disable_http2=disable_http2, disable_http3=disable_http3, disable_ipv4=disable_ipv4, disable_ipv6=disable_ipv6, pool_connections=pool_connections, pool_maxsize=pool_maxsize, happy_eyeballs=happy_eyeballs, keepalive_delay=keepalive_delay, keepalive_idle_window=keepalive_idle_window, revocation_configuration=revocation_configuration, ), )
            self.mount("https://", Adapter(max_retries=retries, resolver=resolver, source_address=source_address, disable_http1=disable_http1, disable_http2=disable_http2, disable_http3=disable_http3, disable_ipv4=disable_ipv4, disable_ipv6=disable_ipv6, pool_connections=pool_connections, pool_maxsize=pool_maxsize, happy_eyeballs=happy_eyeballs, keepalive_delay=keepalive_delay, keepalive_idle_window=keepalive_idle_window, revocation_configuration=revocation_configuration, ), )
        else:
            wasm_adapter = PyodideAdapter(max_retries=retries,)
            for supported_scheme in ["http", "https", "sse", "psse", "ws", "wss"]:
                self.mount(f"{supported_scheme}://", wasm_adapter)
        if app is not None:
            if hasattr(app, "__call__") and iscoroutinefunction(app.__call__):
                adapter = ThreadAsyncServerGatewayInterface(
                    app=app,  # type: ignore[arg-type]
                    max_retries=retries,
                )
                self.mount("asgi://default", adapter)
                for _scheme in ("ws", "wss", "sse", "psse"):
                    self.mount(f"{_scheme}://default", adapter)
                if self.base_url is None:
                    self.base_url = "asgi://default"
            else:
                adapter = WebServerGatewayInterface(  # type: ignore[assignment]
                    app=app,  # type: ignore[arg-type]
                    max_retries=retries,
                )
                self.mount("wsgi://default", adapter)
                for _scheme in ("ws", "wss", "sse", "psse"):
                    self.mount(f"{_scheme}://default", adapter)
                if self.base_url is None:
                    self.base_url = "wsgi://default"
```

--------------------------------

### Send GET Request

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/sessions.html

Sends a GET request to the specified URL. This method is a convenience wrapper for making GET requests with various optional parameters.

```python
    def get(
        self,
        url: str,
        *,
        params: QueryParameterType | None = None,
        headers: HeadersType | None = None,
        cookies: CookiesType | None = None,
        auth: HttpAuthenticationType | None = None,
        timeout: TimeoutType | None = None,
        allow_redirects: bool = True,
        proxies: ProxyType | None = None,
        hooks: HookType[PreparedRequest | Response] | None = None,
        verify: TLSVerifyType | None = None,
        stream: bool | None = None,
        cert: TLSClientCertType | None = None,
        **kwargs: typing.Any,
    ) -> Response:
        r"""Sends a GET request. Returns :class:`Response` object.

        :param url: URL for the new :class:`Request` object.
        :param params: (optional) Dictionary or bytes to be sent in the query
            string for the :class:`Request`.
        :param headers: (optional) Dictionary of HTTP Headers to send with the
            :class:`Request`.
        :param cookies: (optional) Dict or CookieJar object to send with the
            :class:`Request`.

```

--------------------------------

### get

Source: https://niquests.readthedocs.io/en/latest/api.html

Sends a GET request. Returns Response object.

```APIDOC
## GET /<url>

### Description
Sends a GET request. Returns `Response` object.

### Method
GET

### Endpoint
`/<url>`

### Parameters
#### Query Parameters
- **params** (QueryParameterType | None) - Optional - Dictionary or bytes to be sent in the query string for the `Request`.
- **headers** (HeadersType | None) - Optional - Dictionary of HTTP Headers to send with the `Request`.
- **cookies** (CookiesType | None) - Optional - Dict or CookieJar object to send with the `Request`.
- **auth** (HttpAuthenticationType | None) - Optional - Auth tuple or callable to enable Basic/Digest/Custom HTTP Auth.
- **timeout** (TimeoutType | None) - Optional - How long to wait for the server to send data before giving up, as a float, or a (connect timeout, read timeout) tuple.
- **allow_redirects** (bool) - Optional - Set to True by default.
- **proxies** (ProxyType | None) - Optional - Dictionary mapping protocol or protocol and hostname to the URL of the proxy.
- **hooks** (HookType[PreparedRequest | Response] | None) - Optional - Dictionary mapping hook name to one event or list of events, event must be callable.
- **verify** (TLSVerifyType | None) - Optional - Either a boolean, in which case it controls whether we verify the server’s TLS certificate, or a path passed as a string or os.Pathlike object, in which case it must be a path to a CA bundle to use. Defaults to `True`. When set to `False`, requests will accept any TLS certificate presented by the server, and will ignore hostname mismatches and/or expired certificates, which will make your application vulnerable to man-in-the-middle (MitM) attacks. Setting verify to `False` may be useful during local development or testing. It is also possible to put the certificates (directly) in a string or bytes.
- **stream** (bool | None) - Optional - whether to immediately download the response content. Defaults to `False`.
- **cert** (TLSClientCertType | None) - Optional - if String, path to ssl client cert file (.pem). If Tuple, (‘cert’, ‘key’) pair, or (‘cert’, ‘key’, ‘key_password’).

### Response
#### Success Response (200)
- **Response** (Response) - The Response object.

### Request Example
```python
requests.get(
    url='http://example.com',
    params={'key': 'value'},
    headers={'User-Agent': 'my-app'}
)
```

### Response Example
```python
# Response object details would be here
```
```

--------------------------------

### Demonstrate HTTP/2 Multiplexing with niquests

Source: https://niquests.readthedocs.io/en/latest/community/faq.html

This example shows how to compare standard requests with multiplexed requests using niquests to observe performance differences. Adjust REQUEST_COUNT to test various loads.

```python
from time import time
from niquests import Session

#: You can adjust it as you want and verify the multiplexed advantage!
REQUEST_COUNT = 10
REQUEST_URL = "https://httpbin.org/delay/1"

def make_requests(url: str, count: int, use_multiplexed: bool):
  before = time()

  responses = []

  with Session(multiplexed=use_multiplexed) as s:
    for _ in range(count):
      responses.append(s.get(url))
      print(f"request {_+1}...OK")
    print([r.status_code for r in responses])

  print(
      f"{time() - before} seconds elapsed ({'multiplexed' if use_multiplexed else 'standard'})"
  )

#: Let's start with the same good old request one request at a time.
print("> Without multiplexing:")
make_requests(REQUEST_URL, REQUEST_COUNT, False)
#: Now we'll take advantage of a multiplexed connection.
print("> With multiplexing:")
make_requests(REQUEST_URL, REQUEST_COUNT, True)

```

--------------------------------

### Check HTTP/3 and QUIC Support

Source: https://niquests.readthedocs.io/en/latest/user/quickstart.html

Verify if your environment supports HTTP/3 and QUIC. This functionality relies on the `qh3` package.

```python
import niquests

# Check if HTTP/3 is supported by running this command in your terminal:
# python -m niquests.help
```

--------------------------------

### Create and Use an AsyncSession

Source: https://niquests.readthedocs.io/en/latest/api.html

Demonstrates how to create an asynchronous session using niquests and make a GET request. Can be used as a context manager for automatic resource management.

```python
>>> import niquests
>>> s = niquests.AsyncSession()
>>> await s.get('https://httpbin.org/get')
<Response HTTP/2 [200]>
```

```python
>>> async with niquests.AsyncSession() as s:
...     await s.get('https://httpbin.org/get')
<Response HTTP/2 [200]>
```

--------------------------------

### Basic SSE Stream Management (Async)

Source: https://niquests.readthedocs.io/en/latest/user/quickstart.html

This asynchronous example demonstrates consuming Server-Sent Events using Niquests. It utilizes an AsyncSession and requires the 'sse://' scheme.

```python
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())


```

--------------------------------

### Configure UV for urllib3-future cohabitation

Source: https://niquests.readthedocs.io/en/latest/community/faq.html

Configure UV in pyproject.toml to prevent urllib3-future from being installed as a binary, then install Niquests.

```toml
[tool.uv]
no-binary-package = ["urllib3-future"]
```

```bash
$ export URLLIB3_NO_OVERRIDE=1
$ uv add niquests
```

--------------------------------

### Example: Custom Async Middleware

Source: https://niquests.readthedocs.io/en/latest/user/advanced.html

Demonstrates how to create and use a custom asynchronous middleware by subclassing `AsyncLifeCycleHook` and overriding the `pre_send` method to log connection information.

```python
import asyncio
from niquests import AsyncSession, AsyncLifeCycleHook

class ConnectionLogger(AsyncLifeCycleHook):
    async def pre_send(self, prepared_request, **kwargs) -> None:
        # Inspect the connection info before the request is sent
        print(f"Connected to: {prepared_request.conn_info}")

async def main():
    async with AsyncSession() as s:
        # Pass an instance of your hook class
        await s.get("https://one.one.one.one", hooks=ConnectionLogger())

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

--------------------------------

### Configure PDM for urllib3-future cohabitation

Source: https://niquests.readthedocs.io/en/latest/community/faq.html

Configure PDM in pyproject.toml to prevent urllib3-future from being installed as a binary, then install Niquests.

```toml
[tool.pdm.resolution]
no-binary = "urllib3-future"
```

```bash
$ export URLLIB3_NO_OVERRIDE=1
$ pdm add niquests
```

--------------------------------

### Async GET Request

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/async_session.html

Makes an asynchronous GET request to the specified URL. Supports various parameters for customization.

```APIDOC
## GET /url

### Description
Makes an asynchronous GET request to the specified URL. Supports various parameters for customization.

### Method
GET

### Endpoint
/{url}

### Parameters
#### Query Parameters
- **params** (QueryParameterType | None) - Optional - Query parameters for the request.
- **headers** (HeadersType | None) - Optional - Custom headers for the request.
- **cookies** (CookiesType | None) - Optional - Cookies to send with the request.
- **auth** (HttpAuthenticationType | AsyncHttpAuthenticationType | None) - Optional - Authentication method for the request.
- **timeout** (TimeoutType | None) - Optional - Request timeout duration.
- **allow_redirects** (bool) - Optional - Whether to follow redirects (default: True).
- **proxies** (ProxyType | None) - Optional - Proxy settings for the request.
- **hooks** (AsyncHookType[PreparedRequest | Response] | None) - Optional - Hooks to run before/after the request.
- **verify** (TLSVerifyType | None) - Optional - SSL certificate verification settings.
- **stream** (bool | None) - Optional - Whether to stream the response.
- **cert** (TLSClientCertType | None) - Optional - Client certificate for the request.

### Request Example
```python
async with AsyncSession() as session:
    response = await session.get("https://example.com", params={"key": "value"})
```

### Response
#### Success Response (200)
- **Response** (Response) - The response object if stream is False or None.
- **AsyncResponse** (AsyncResponse) - The asynchronous response object if stream is True.
```

--------------------------------

### Install Niquests with urllib3-future (Poetry Option 1)

Source: https://niquests.readthedocs.io/en/latest/community/faq.html

Configure Poetry to not use binary wheels for urllib3-future and then add Niquests. This ensures Niquests uses the correct urllib3 version.

```bash
$ export URLLIB3_NO_OVERRIDE=1
$ poetry config --local installer.no-binary urllib3-future
$ poetry add niquests
```

--------------------------------

### Create Niquests Session Instance

Source: https://niquests.readthedocs.io/en/latest/dev/httpx.html

Migrate from HTTPX's `Client` to Niquests' `Session`. Note that not all keyword arguments supported by `httpx.Client` may exist on `niquests.Session`.

```python
session = niquests.Session(**kwargs)
```

--------------------------------

### Make a GET Request (Async)

Source: https://niquests.readthedocs.io/en/latest/user/quickstart.html

Make an asynchronous GET request to fetch data from a URL. Use 'await' for non-blocking operations.

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

```

--------------------------------

### Prepare and Send a POST Request (Sync)

Source: https://niquests.readthedocs.io/en/latest/user/advanced.html

Prepare a request object manually, modify its body and headers, and then send it using a synchronous session. This allows for pre-request modifications.

```python
from niquests import Request, Session

s = Session()

req = Request('POST', url, data=data, headers=headers)
prepped = req.prepare()

# do something with prepped.body
prepped.body = 'No, I want exactly this as the body.'

# do something with prepped.headers
del prepped.headers['Content-Type']

resp = s.send(prepped,
    stream=stream,
    verify=verify,
    proxies=proxies,
    cert=cert,
    timeout=timeout
)

print(resp.status_code)
```

--------------------------------

### Make a GET Request (Sync)

Source: https://niquests.readthedocs.io/en/latest/user/quickstart.html

Make a synchronous GET request to fetch data from a URL. The response is stored in the 'r' object.

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

```

--------------------------------

### Initialize a Request Object

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/models.html

Demonstrates the initialization of a Request object with various parameters like method, URL, headers, and data. Default empty dictionaries are used for parameters like data and headers if not provided.

```python
def __init__(
        self,
        method: HttpMethodType | None = None,
        url: str | None = None,
        headers: HeadersType | None = None,
        files: MultiPartFilesType | MultiPartFilesAltType | None = None,
        data: BodyType | AsyncBodyType | None = None,
        params: QueryParameterType | None = None,
        auth: HttpAuthenticationType | AsyncHttpAuthenticationType | None = None,
        cookies: CookiesType | None = None,
        hooks: HookType | AsyncHookType | None = None,
        json: typing.Any | None = None,
        base_url: str | None = None,
    ):
        # Default empty dicts for dict params.
        data = [] if data is None else data
        files = [] if files is None else files
        headers = CaseInsensitiveDict() if headers is None else headers
        params = {} if params is None else params
        hooks = {} if hooks is None else hooks

        self.hooks: HookType[Response | PreparedRequest] = default_hooks()
        for k, v in list(hooks.items()):
            self.register_hook(event=k, hook=v)

        self.method = method
        self.url = url
        self.headers = headers
        self.files = files
        self.data = data
        self.json = json
        self.params = params
        self.auth = auth
        self.cookies = cookies
        self.base_url = base_url
```

--------------------------------

### WSGI Testing with httpx

Source: https://niquests.readthedocs.io/en/latest/dev/httpx.html

Demonstrates how to test a WSGI application using httpx's WSGITransport. This setup is useful for testing synchronous web frameworks.

```python
from flask import Flask
import httpx


app = Flask(__name__)

@app.route("/")
def hello():
    return "Hello World!"

transport = httpx.WSGITransport(app=app)
with httpx.Client(transport=transport, base_url="http://testserver") as client:
    r = client.get("/")
    assert r.status_code == 200
    assert r.text == "Hello World!"

```

--------------------------------

### GET Request

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/api.html

Sends a GET request to the specified URL. This method does not keep the connection alive; use a Session object to reuse connections.

```APIDOC
## GET /url

### Description
Sends a GET request to the specified URL. This method does not keep the connection alive. Use a :class:`Session` to reuse the connection.

### Method
GET

### Endpoint
/{url}

### Parameters
#### Path Parameters
- **url** (str) - Required - The URL for the new :class:`Request` object.

#### Query Parameters
- **params** (QueryParameterType | None) - Optional - Dictionary, list of tuples or bytes to send in the query string for the :class:`Request`.

#### Headers
- **headers** (HeadersType | None) - Optional - Dictionary of HTTP Headers to send with the :class:`Request`.

#### Cookies
- **cookies** (CookiesType | None) - Optional - Dict or CookieJar object to send with the :class:`Request`.

#### Authentication
- **auth** (HttpAuthenticationType | None) - Optional - Auth tuple to enable Basic/Digest/Custom HTTP Auth.

#### Timeout
- **timeout** (TimeoutType | None) - Optional - How many seconds to wait for the server to send data before giving up, as a float, or a :ref:`(connect timeout, read timeout) <timeouts>` tuple. Defaults to READ_DEFAULT_TIMEOUT.

#### Redirects
- **allow_redirects** (bool) - Optional - Enable/disable GET/OPTIONS/POST/PUT/PATCH/DELETE/HEAD redirection. Defaults to ``True``.

#### Proxies
- **proxies** (ProxyType | None) - Optional - Dictionary mapping protocol to the URL of the proxy.

#### Verification
- **verify** (TLSVerifyType) - Optional - Either a boolean, in which case it controls whether we verify the server's TLS certificate, or a path passed as a string or os.Pathlike object, in which case it must be a path to a CA bundle to use. Defaults to ``True``. It is also possible to put the certificates (directly) in a string or bytes.

#### Stream
- **stream** (bool) - Optional - If ``False``, the response content will be immediately downloaded. Defaults to ``False``.

#### Certificate
- **cert** (TLSClientCertType | None) - Optional - If String, path to ssl client cert file (.pem). If Tuple, ('cert', 'key') pair, or ('cert', 'key', 'key_password').

#### Hooks
- **hooks** (HookType[PreparedRequest | Response] | None) - Optional - Register functions that should be called at very specific moments in the request lifecycle.

#### Retries
- **retries** (RetryType) - Optional - If integer, determine the number of retry in case of a timeout or connection error. Otherwise, for fine gained retry, use directly a ``Retry`` instance from urllib3. Defaults to DEFAULT_RETRIES.

### Return Value
- :class:`Response <Response>` object

```

--------------------------------

### ASGI Testing with Niquests (AsyncSession)

Source: https://niquests.readthedocs.io/en/latest/dev/httpx.html

Illustrates how to test ASGI applications using Niquests' AsyncSession. Note that ASGI lifespan startup/shutdown is not handled and may require external libraries like asgi-lifespan.

```python
from niquests import AsyncSession

# Assuming 'app' is your ASGI application instance
# with niquests.AsyncSession(app=app) as s:
#     r = await s.get("/")

```

--------------------------------

### Niquests Session GET Request

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/sessions.html

Makes a GET request to a specified URL with various optional parameters for customization. Use this for retrieving data from a server.

```python
def get(
    self,
    url: str,
    params: QueryParameterType | None = None,
    headers: HeadersType | None = None,
    cookies: CookiesType | None = None,
    auth: HttpAuthenticationType | None = None,
    timeout: TimeoutType | None = None,
    allow_redirects: bool = True,
    proxies: ProxyType | None = None,
    hooks: HookType[PreparedRequest | Response] | None = None,
    verify: TLSVerifyType | None = None,
    stream: bool | None = None,
    cert: TLSClientCertType | None = None,
    **kwargs: typing.Any,
) -> Response:
    r"""Sends a GET request. Returns :class:`Response` object.

    :param url: URL for the new :class:`Request` object.
    :param params: (optional) Dictionary or bytes to be sent in the query
        string for the :class:`Request`.
    :param headers: (optional) Dictionary of HTTP Headers to send with the
        :class:`Request`.
    :param cookies: (optional) Dict or CookieJar object to send with the
        :class:`Request`.
    :param auth: (optional) Auth tuple or callable to enable
        Basic/Digest/Custom HTTP Auth.
    :param timeout: (optional) How long to wait for the server to send
        data before giving up, as a float, or a :ref:`(connect timeout,
        read timeout) <timeouts>` tuple.
    :param allow_redirects: (optional) Set to True by default.
    :param proxies: (optional) Dictionary mapping protocol or protocol and
        hostname to the URL of the proxy.
    :param hooks: (optional) Dictionary mapping hook name to one event or
        list of events, event must be callable.
    :param stream: (optional) whether to immediately download the response
        content. Defaults to ``False``.
    :param verify: (optional) Either a boolean, in which case it controls whether we verify
        the server's TLS certificate, or a path passed as a string or os.Pathlike object,
        in which case it must be a path to a CA bundle to use.
        Defaults to ``True``.
        When set to
        ``False``, requests will accept any TLS certificate presented by
        the server, and will ignore hostname mismatches and/or expired
        certificates, which will make your application vulnerable to
        man-in-the-middle (MitM) attacks. Setting verify to ``False``
        may be useful during local development or testing.
        It is also possible to put the certificates (directly) in a string or bytes.
    :param cert: (optional) if String, path to ssl client cert file (.pem).
        If Tuple, ('cert', 'key') pair, or ('cert', 'key', 'key_password').
    """

    return self.request(
        "GET",
        url,
        params=params,
        headers=headers,
        cookies=cookies,
        auth=auth,
        timeout=timeout,
        allow_redirects=allow_redirects,
        proxies=proxies,
        hooks=hooks,
        verify=verify,
        stream=stream,
        cert=cert,
        **kwargs,
    )

```

--------------------------------

### RequestsCookieJar get() Method

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/cookies.html

Provides a dictionary-like `get()` method for `RequestsCookieJar`. It supports optional domain and path arguments to resolve naming collisions, but this operation is O(n).

```python
def get(self, name, default=None, domain=None, path=None):
        """Dict-like get() that also supports optional domain and path args in
        order to resolve naming collisions from using one cookie jar over
        multiple domains.

        .. warning:: operation is O(n), not O(1).
        """
        try:
            return self._find_no_duplicates(name, domain, path)
        except KeyError:
            return default
```

--------------------------------

### Perform an HTTP GET Request

Source: https://niquests.readthedocs.io/en/latest/user/advanced.html

Use the GET verb to retrieve a resource from a URL. This is idempotent and suitable for fetching data. Ensure the status code indicates success before processing the response.

```python
>>> import niquests
>>> r = niquests.get('https://api.github.com/repos/psf/requests/git/commits/a050faf084662f3a352dd1a941f2c7c9f886d4ad')
```

--------------------------------

### AsyncHTTPAdapter Initialization

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/adapters.html

Initializes the AsyncHTTPAdapter with various configuration options for asynchronous HTTP requests. This includes settings for connection pooling, retry mechanisms, and HTTP protocol versions.

```APIDOC
## AsyncHTTPAdapter(pool_connections=DEFAULT_POOLSIZE, pool_maxsize=DEFAULT_POOLSIZE, max_retries=DEFAULT_RETRIES, pool_block=DEFAULT_POOLBLOCK, *, quic_cache_layer=None, disable_http1=False, disable_http2=False, disable_http3=False, max_in_flight_multiplexed=None, resolver=None, source_address=None, disable_ipv4=False, disable_ipv6=False, happy_eyeballs=False, keepalive_delay=600.0, keepalive_idle_window=60.0, revocation_configuration=DEFAULT_STRATEGY)

### Description
Initializes the AsyncHTTPAdapter with various configuration options for asynchronous HTTP requests. This includes settings for connection pooling, retry mechanisms, and HTTP protocol versions.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
import niquests

s = niquests.AsyncSession()
a = niquests.adapters.AsyncHTTPAdapter(max_retries=3)
s.mount('http://', a)
```

### Response
#### Success Response (200)
None

#### Response Example
None
```

--------------------------------

### Niquests GET Request

Source: https://niquests.readthedocs.io/en/latest/_modules/niquests/api.html

Sends a GET request to the specified URL. Use a Session object to reuse the connection. Supports various parameters like headers, cookies, authentication, and timeouts.

```python
def get(
    url: str,
    params: QueryParameterType | None = None,
    *,
    headers: HeadersType | None = None,
    cookies: CookiesType | None = None,
    auth: HttpAuthenticationType | None = None,
    timeout: TimeoutType | None = READ_DEFAULT_TIMEOUT,
    allow_redirects: bool = True,
    proxies: ProxyType | None = None,
    verify: TLSVerifyType = True,
    stream: bool = False,
    cert: TLSClientCertType | None = None,
    hooks: HookType[PreparedRequest | Response] | None = None,
    retries: RetryType = DEFAULT_RETRIES,
    **kwargs: typing.Any,
) -> Response:
    r"""Sends a GET request. This does not keep the connection alive. Use a :class:`Session` to reuse the connection.

    :param url: URL for the new :class:`Request` object.
    :param params: (optional) Dictionary, list of tuples or bytes to send
        in the query string for the :class:`Request`.
    :param params: (optional) Dictionary, list of tuples or bytes to send
        in the query string for the :class:`Request`.
    :param headers: (optional) Dictionary of HTTP Headers to send with the :class:`Request`.
    :param cookies: (optional) Dict or CookieJar object to send with the :class:`Request`.
    :param auth: (optional) Auth tuple to enable Basic/Digest/Custom HTTP Auth.
    :param timeout: (optional) How many seconds to wait for the server to send data
        before giving up, as a float, or a :ref:`(connect timeout, read
        timeout) <timeouts>` tuple.
    :param allow_redirects: (optional) Boolean. Enable/disable GET/OPTIONS/POST/PUT/PATCH/DELETE/HEAD redirection.
            Defaults to ``True``.
    :param proxies: (optional) Dictionary mapping protocol to the URL of the proxy.
    :param verify: (optional) Either a boolean, in which case it controls whether we verify
            the server's TLS certificate, or a path passed as a string or os.Pathlike object,
            in which case it must be a path to a CA bundle to use.
            Defaults to ``True``.
            It is also possible to put the certificates (directly) in a string or bytes.
    :param stream: (optional) if ``False``, the response content will be immediately downloaded.
    :param cert: (optional) if String, path to ssl client cert file (.pem).
            If Tuple, ('cert', 'key') pair, or ('cert', 'key', 'key_password').
    :param hooks: (optional) Register functions that should be called at very specific moment in the request lifecycle.
    :param retries: (optional) If integer, determine the number of retry in case of a timeout or connection error.
            Otherwise, for fine gained retry, use directly a ``Retry`` instance from urllib3.
    :return: :class:`Response <Response>` object
    """

    return request(
        "GET",
        url,
        params=params,
        headers=headers,
        cookies=cookies,
        auth=auth,
        timeout=timeout,
        allow_redirects=allow_redirects,
        proxies=proxies,
        verify=verify,
        stream=stream,
        cert=cert,
        hooks=hooks,
        retries=retries,
        **kwargs,
    )
```
Tokens:
Raw