or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

advanced-features.mdasync-requests.mdexceptions.mdindex.mdmodels.mdsessions.mdsync-requests.md
tile.json

tessl/pypi-niquests

Niquests is a simple, yet elegant, HTTP library that is a drop-in replacement for Requests, which is under feature freeze.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/niquests@3.15.x

To install, run

npx @tessl/cli install tessl/pypi-niquests@3.15.0

index.mddocs/

Niquests

Niquests is a modern, high-performance HTTP client library for Python that serves as a drop-in replacement for the popular Requests library. It offers advanced features including HTTP/2 and HTTP/3 support with multiplexing capabilities, asynchronous and synchronous APIs, DNS over HTTPS/QUIC/TLS, OCSP certificate revocation verification, WebSocket support, and comprehensive network fine-tuning options.

Package Information

  • Package Name: niquests
  • Package Type: pypi
  • Language: Python
  • Installation: pip install niquests

Core Imports

import niquests

Common usage patterns:

# Synchronous requests
from niquests import get, post, Session

# Asynchronous requests
from niquests import aget, apost, AsyncSession

# Models and exceptions
from niquests import Response, Request, RequestException

Basic Usage

import niquests

# Simple GET request
response = niquests.get('https://httpbin.org/get')
print(response.status_code)
print(response.json())

# POST request with data
data = {'key': 'value'}
response = niquests.post('https://httpbin.org/post', json=data)
print(response.json())

# Using sessions for persistent connections
with niquests.Session() as session:
    response = session.get('https://httpbin.org/get')
    print(response.text)

# Async requests
import asyncio

async def fetch_data():
    response = await niquests.aget('https://httpbin.org/get')
    return response.json()

# Run async function
data = asyncio.run(fetch_data())

Architecture

Niquests maintains compatibility with the Requests API while providing significant performance improvements through:

  • HTTP/2 and HTTP/3 Support: Modern protocol implementations with multiplexing
  • Dual API Design: Both synchronous and asynchronous interfaces
  • Connection Pooling: Automatic connection reuse and management
  • Advanced Security: OCSP verification, post-quantum cryptography, DNS over secure channels
  • Drop-in Compatibility: Full API compatibility with existing Requests code

The library is built on top of urllib3 with extensive enhancements for modern networking requirements.

Capabilities

Synchronous HTTP Requests

Core HTTP methods for making synchronous requests including GET, POST, PUT, PATCH, DELETE, HEAD, and OPTIONS. These functions provide a simple interface for HTTP communication without requiring explicit session management.

def request(method: HttpMethodType, url: str, *, 
            params: QueryParameterType | None = None,
            data: BodyType | None = None,
            json: Any | None = None,
            headers: HeadersType | None = None,
            cookies: CookiesType | None = None,
            files: MultiPartFilesType | MultiPartFilesAltType | None = None,
            auth: HttpAuthenticationType | None = None,
            timeout: TimeoutType | None = WRITE_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) -> Response: ...

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: Any) -> Response: ...

def post(url: str, data: BodyType | None = None, json: Any | None = None, *, 
         params: QueryParameterType | None = None,
         headers: HeadersType | None = None,
         cookies: CookiesType | None = None,
         files: MultiPartFilesType | MultiPartFilesAltType | None = None,
         auth: HttpAuthenticationType | None = None,
         timeout: TimeoutType | None = WRITE_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) -> Response: ...

def put(url: str, data: BodyType | None = None, *, 
        json: Any | None = None,
        params: QueryParameterType | None = None,
        headers: HeadersType | None = None,
        cookies: CookiesType | None = None,
        files: MultiPartFilesType | MultiPartFilesAltType | None = None,
        auth: HttpAuthenticationType | None = None,
        timeout: TimeoutType | None = WRITE_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) -> Response: ...

def patch(url: str, data: BodyType | None = None, *, 
          json: Any | None = None,
          params: QueryParameterType | None = None,
          headers: HeadersType | None = None,
          cookies: CookiesType | None = None,
          files: MultiPartFilesType | MultiPartFilesAltType | None = None,
          auth: HttpAuthenticationType | None = None,
          timeout: TimeoutType | None = WRITE_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) -> Response: ...

def delete(url: str, *, 
           params: QueryParameterType | None = None,
           headers: HeadersType | None = None,
           cookies: CookiesType | None = None,
           auth: HttpAuthenticationType | None = None,
           timeout: TimeoutType | None = WRITE_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: Any) -> Response: ...

def head(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 = False,
         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: Any) -> Response: ...

def options(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: Any) -> Response: ...

Synchronous Requests

Asynchronous HTTP Requests

Async versions of all HTTP methods for high-performance concurrent request handling. These functions enable non-blocking HTTP operations and are ideal for applications requiring high throughput. Return type depends on stream parameter.

async def arequest(method: HttpMethodType, url: str, *, 
                   params: QueryParameterType | None = None,
                   data: BodyType | AsyncBodyType | None = None,
                   headers: HeadersType | None = None,
                   cookies: CookiesType | None = None,
                   files: MultiPartFilesType | MultiPartFilesAltType | None = None,
                   auth: HttpAuthenticationType | AsyncHttpAuthenticationType | None = None,
                   timeout: TimeoutType | None = WRITE_DEFAULT_TIMEOUT,
                   allow_redirects: bool = True,
                   proxies: ProxyType | None = None,
                   hooks: AsyncHookType[PreparedRequest | Response] | None = None,
                   stream: bool | None = None,
                   verify: TLSVerifyType | None = None,
                   cert: TLSClientCertType | None = None,
                   json: Any | None = None,
                   retries: RetryType = DEFAULT_RETRIES) -> Response | AsyncResponse: ...

async def aget(url: str, params: QueryParameterType | None = None, *, 
               headers: HeadersType | None = None,
               cookies: CookiesType | None = None,
               auth: HttpAuthenticationType | AsyncHttpAuthenticationType | None = None,
               timeout: TimeoutType | None = READ_DEFAULT_TIMEOUT,
               allow_redirects: bool = True,
               proxies: ProxyType | None = None,
               hooks: AsyncHookType[PreparedRequest | Response] | None = None,
               verify: TLSVerifyType | None = None,
               stream: bool | None = None,
               cert: TLSClientCertType | None = None,
               retries: RetryType = DEFAULT_RETRIES,
               **kwargs: Any) -> Response | AsyncResponse: ...

async def apost(url: str, data: BodyType | AsyncBodyType | None = None, json: Any | None = None, *, 
                params: QueryParameterType | None = None,
                headers: HeadersType | None = None,
                cookies: CookiesType | None = None,
                files: MultiPartFilesType | MultiPartFilesAltType | None = None,
                auth: HttpAuthenticationType | AsyncHttpAuthenticationType | None = None,
                timeout: TimeoutType | None = WRITE_DEFAULT_TIMEOUT,
                allow_redirects: bool = True,
                proxies: ProxyType | None = None,
                hooks: AsyncHookType[PreparedRequest | Response] | None = None,
                verify: TLSVerifyType | None = None,
                stream: bool | None = None,
                cert: TLSClientCertType | None = None,
                retries: RetryType = DEFAULT_RETRIES) -> Response | AsyncResponse: ...

async def aput(url: str, data: BodyType | AsyncBodyType | None = None, *, 
               json: Any | None = None,
               params: QueryParameterType | None = None,
               headers: HeadersType | None = None,
               cookies: CookiesType | None = None,
               files: MultiPartFilesType | MultiPartFilesAltType | None = None,
               auth: HttpAuthenticationType | AsyncHttpAuthenticationType | None = None,
               timeout: TimeoutType | None = WRITE_DEFAULT_TIMEOUT,
               allow_redirects: bool = True,
               proxies: ProxyType | None = None,
               hooks: AsyncHookType[PreparedRequest | Response] | None = None,
               verify: TLSVerifyType | None = None,
               stream: bool | None = None,
               cert: TLSClientCertType | None = None,
               retries: RetryType = DEFAULT_RETRIES) -> Response | AsyncResponse: ...

async def apatch(url: str, data: BodyType | AsyncBodyType | None = None, *, 
                 json: Any | None = None,
                 params: QueryParameterType | None = None,
                 headers: HeadersType | None = None,
                 cookies: CookiesType | None = None,
                 files: MultiPartFilesType | MultiPartFilesAltType | None = None,
                 auth: HttpAuthenticationType | AsyncHttpAuthenticationType | None = None,
                 timeout: TimeoutType | None = WRITE_DEFAULT_TIMEOUT,
                 allow_redirects: bool = True,
                 proxies: ProxyType | None = None,
                 hooks: AsyncHookType[PreparedRequest | Response] | None = None,
                 verify: TLSVerifyType | None = None,
                 stream: bool | None = None,
                 cert: TLSClientCertType | None = None,
                 retries: RetryType = DEFAULT_RETRIES) -> Response | AsyncResponse: ...

async def adelete(url: str, *, 
                  params: QueryParameterType | None = None,
                  headers: HeadersType | None = None,
                  cookies: CookiesType | None = None,
                  auth: HttpAuthenticationType | AsyncHttpAuthenticationType | None = None,
                  timeout: TimeoutType | None = WRITE_DEFAULT_TIMEOUT,
                  allow_redirects: bool = True,
                  proxies: ProxyType | None = None,
                  hooks: AsyncHookType[PreparedRequest | Response] | None = None,
                  verify: TLSVerifyType | None = None,
                  stream: bool | None = None,
                  cert: TLSClientCertType | None = None,
                  retries: RetryType = DEFAULT_RETRIES,
                  **kwargs: Any) -> Response | AsyncResponse: ...

async def ahead(url: str, *, 
                params: QueryParameterType | None = None,
                headers: HeadersType | None = None,
                cookies: CookiesType | None = None,
                auth: HttpAuthenticationType | AsyncHttpAuthenticationType | None = None,
                timeout: TimeoutType | None = READ_DEFAULT_TIMEOUT,
                allow_redirects: bool = False,
                proxies: ProxyType | None = None,
                hooks: AsyncHookType[PreparedRequest | Response] | None = None,
                verify: TLSVerifyType | None = None,
                stream: bool | None = None,
                cert: TLSClientCertType | None = None,
                retries: RetryType = DEFAULT_RETRIES,
                **kwargs: Any) -> Response | AsyncResponse: ...

async def aoptions(url: str, *, 
                   params: QueryParameterType | None = None,
                   headers: HeadersType | None = None,
                   cookies: CookiesType | None = None,
                   auth: HttpAuthenticationType | AsyncHttpAuthenticationType | None = None,
                   timeout: TimeoutType | None = READ_DEFAULT_TIMEOUT,
                   allow_redirects: bool = True,
                   proxies: ProxyType | None = None,
                   hooks: AsyncHookType[PreparedRequest | Response] | None = None,
                   verify: TLSVerifyType | None = None,
                   stream: bool | None = None,
                   cert: TLSClientCertType | None = None,
                   retries: RetryType = DEFAULT_RETRIES,
                   **kwargs: Any) -> Response | AsyncResponse: ...

Asynchronous Requests

Session Management

Session classes for managing persistent connections, cookies, authentication, and other settings across multiple requests. Sessions provide connection pooling and configuration management.

class Session:
    def request(self, method: HttpMethodType, url: str, **kwargs) -> Response: ...
    def get(self, url: str, **kwargs) -> Response: ...
    def post(self, url: str, **kwargs) -> Response: ...

class AsyncSession:
    async def request(self, method: HttpMethodType, url: str, **kwargs) -> AsyncResponse: ...
    async def get(self, url: str, **kwargs) -> AsyncResponse: ...
    async def post(self, url: str, **kwargs) -> AsyncResponse: ...

Session Management

Request and Response Models

Core data structures for representing HTTP requests and responses. These classes provide access to all HTTP components including headers, body content, status codes, and metadata.

class Request:
    def __init__(self, method: HttpMethodType, url: str, **kwargs): ...

class PreparedRequest:
    method: HttpMethodType
    url: str
    headers: HeadersType

class Response:
    status_code: int
    headers: HeadersType
    content: bytes
    def json(self) -> Any: ...

class AsyncResponse:
    status_code: int
    headers: HeadersType
    content: bytes
    async def json(self) -> Any: ...

Request and Response Models

Exception Handling

Comprehensive exception hierarchy for handling various error conditions during HTTP operations. These exceptions provide detailed error information and enable robust error handling.

class RequestException(IOError): ...
class HTTPError(RequestException): ...
class ConnectionError(RequestException): ...
class Timeout(RequestException): ...
class JSONDecodeError(RequestException): ...

Exception Handling

Advanced Features

Configuration classes, status code utilities, and advanced networking options for fine-tuning HTTP behavior and handling complex scenarios.

class TimeoutConfiguration: ...
class RetryConfiguration: ...

# Status codes lookup
codes: LookupDict

# Legacy compatibility flag
HAS_LEGACY_URLLIB3: bool

Advanced Features

Utilities

Utility functions for HTTP handling, URL processing, authentication, and data conversion.

def super_len(o: Any) -> int: ...
def get_netrc_auth(url: str | None, raise_errors: bool = False) -> tuple[str, str] | None: ...
def guess_filename(obj: IO) -> str | None: ...
def from_key_val_list(value: Any | None) -> OrderedDict | None: ...
def to_key_val_list(value: list | dict | OrderedDict | None) -> list[tuple[str, str]] | None: ...
def parse_list_header(value: str) -> list[str]: ...
def parse_dict_header(value: str) -> Mapping[str, str | None]: ...
def unquote_header_value(value: str, is_filename: bool = False) -> str: ...
def dict_from_cookiejar(cj: CookieJar) -> dict[str, str | None]: ...
def add_dict_to_cookiejar(cj: RequestsCookieJar, cookie_dict) -> RequestsCookieJar | CookieJar: ...
def get_encoding_from_headers(headers: Mapping[str, str]) -> str | None: ...
def stream_decode_response_unicode(iterator: Iterator[bytes], encoding: str) -> Iterator[str]: ...
def iter_slices(string: str, slice_length: int | None) -> Generator[str, None, None]: ...
def unquote_unreserved(uri: str) -> str: ...
def requote_uri(uri: str) -> str: ...
def address_in_network(ip: str, net: str) -> bool: ...
def dotted_netmask(mask: int) -> str: ...

# Utils module reference
utils: ModuleType

Package Metadata

__version__: str
__title__: str
__description__: str
__url__: str
__author__: str
__author_email__: str
__license__: str
__copyright__: str
__build__: str
__cake__: str

Type Definitions

# Import statements
from typing import (
    Any, Union, List, Tuple, Dict, Mapping, MutableMapping, 
    Optional, Callable, Awaitable, Iterable, AsyncIterable, 
    IO, Iterator, Generator, TypeVar
)
from http.cookiejar import CookieJar
from os import PathLike
from collections import OrderedDict
from types import ModuleType

# Type variables
_T = TypeVar('_T')

# Basic type aliases
HttpMethodType = str

# Query parameters and headers
QueryParameterType = Union[
    List[Tuple[str, Union[str, List[str], None]]],
    Mapping[str, Union[str, List[str], None]],
    bytes,
    str,
]

HeadersType = Union[
    MutableMapping[Union[str, bytes], Union[str, bytes]],
    MutableMapping[str, str],
    MutableMapping[bytes, bytes],
    CaseInsensitiveDict,
    List[Tuple[Union[str, bytes], Union[str, bytes]]],
    Headers,
]

# Request body types
BodyFormType = Union[
    List[Tuple[str, str]],
    Dict[str, Union[List[str], str]],
]

BodyType = Union[
    str,
    bytes,
    bytearray,
    IO,
    BodyFormType,
    Iterable[bytes],
    Iterable[str],
]

AsyncBodyType = Union[
    AsyncIterable[bytes],
    AsyncIterable[str],
]

# Cookies and authentication
CookiesType = Union[
    MutableMapping[str, str],
    CookieJar,
]

HttpAuthenticationType = Union[
    Tuple[Union[str, bytes], Union[str, bytes]],
    str,
    AuthBase,
    Callable[[PreparedRequest], PreparedRequest],
]

AsyncHttpAuthenticationType = Union[
    AsyncAuthBase,
    Callable[[PreparedRequest], Awaitable[PreparedRequest]],
]

# TLS and security
TLSVerifyType = Union[bool, str, bytes, PathLike]
TLSClientCertType = Union[str, Tuple[str, str], Tuple[str, str, str]]

# Timeouts and retries
TimeoutType = Union[
    int,
    float,
    Tuple[Union[int, float], Union[int, float]],
    Tuple[Union[int, float], Union[int, float], Union[int, float]],
    Timeout,
]

RetryType = Union[bool, int, Retry]

# Proxies and networking
ProxyType = Dict[str, str]

# File uploads
BodyFileType = Union[str, bytes, bytearray, IO]

MultiPartFileType = Tuple[
    str,
    Union[
        BodyFileType,
        Tuple[str, BodyFileType],
        Tuple[str, BodyFileType, str],
        Tuple[str, BodyFileType, str, HeadersType],
    ],
]

MultiPartFilesType = List[MultiPartFileType]

MultiPartFilesAltType = Dict[
    str,
    Union[
        BodyFileType,
        Tuple[str, BodyFileType],
        Tuple[str, BodyFileType, str],
        Tuple[str, BodyFileType, str, HeadersType],
    ],
]

# Hooks and callbacks
HookCallableType = Callable[[_T], Optional[_T]]
HookType = Dict[str, List[HookCallableType[_T]]]

AsyncHookCallableType = Callable[[_T], Awaitable[Optional[_T]]]
AsyncHookType = Dict[str, List[Union[HookCallableType[_T], AsyncHookCallableType[_T]]]]

# Cache and connection management
CacheLayerAltSvcType = MutableMapping[Tuple[str, int], Optional[Tuple[str, int]]]

# Resolver types for DNS
ResolverType = Union[
    str,
    ResolverDescription,
    BaseResolver,
    List[str],
    List[ResolverDescription],
]

AsyncResolverType = Union[
    str,
    AsyncResolverDescription,
    AsyncBaseResolver,
    List[str],
    List[AsyncResolverDescription],
]

# Default timeout constants
READ_DEFAULT_TIMEOUT: TimeoutType
WRITE_DEFAULT_TIMEOUT: TimeoutType
DEFAULT_RETRIES: RetryType

# Additional exported classes and utilities
class CaseInsensitiveDict: ...
class Headers: ...
class AuthBase: ...
class AsyncAuthBase: ...
class Retry: ...
class Timeout: ...
class LookupDict: ...
class ResolverDescription: ...
class AsyncResolverDescription: ...
class BaseResolver: ...
class AsyncBaseResolver: ...
class RequestsCookieJar(CookieJar): ...