tessl/pypi-niquests

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

—

Pending

Overview

Eval results

Files

Asynchronous HTTP Requests

Name: tessl/pypi-niquests
Author: tessl

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 or concurrent request processing.

All async functions are prefixed with 'a' and return AsyncResponse objects. They support the same parameters as their synchronous counterparts.

Capabilities

Generic Async Request Function

The fundamental async request function that all other async HTTP methods build upon.

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:
    """
    Asynchronously constructs and sends a Request with the specified HTTP method.
    
    Args:
        method: HTTP method (GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS)
        url: Target URL for the request
        params: Query parameters to append to URL
        data: Request body data (supports async iterables)
        json: JSON serializable object to send in request body
        headers: HTTP headers to include in request
        cookies: Cookies to send with request
        files: Files for multipart upload
        auth: Authentication credentials (supports async auth)
        timeout: Request timeout in seconds
        allow_redirects: Whether to follow HTTP redirects
        proxies: Proxy configuration
        verify: SSL certificate verification
        stream: Controls response loading (None/False=immediate download as Response, True=lazy loading as AsyncResponse)
        cert: Client SSL certificate
        hooks: Request/response lifecycle hooks
        retries: Retry configuration
        
    Returns:
        Response object if stream=None/False, AsyncResponse object if stream=True
    """

Async GET Requests

Asynchronously retrieves data from the specified URL.

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:
    """
    Asynchronously sends a GET request to retrieve data.
    
    Args:
        url: Target URL
        params: Query parameters
        headers: HTTP headers
        cookies: Cookies to send
        auth: Authentication credentials
        timeout: Request timeout (defaults to read timeout)
        allow_redirects: Follow redirects (default: True)
        proxies: Proxy configuration
        verify: SSL verification
        stream: Stream response content
        cert: Client certificate
        hooks: Request hooks
        retries: Retry configuration
        **kwargs: Additional arguments
        
    Returns:
        Response object if stream=None/False, AsyncResponse object if stream=True
    """

Usage example:

import asyncio
import niquests

async def fetch_data():
    # Simple async GET request
    response = await niquests.aget('https://api.example.com/data')
    data = await response.json()
    return data

# Concurrent requests
async def fetch_multiple():
    urls = [
        'https://api.example.com/users/1',
        'https://api.example.com/users/2',
        'https://api.example.com/users/3'
    ]
    
    tasks = [niquests.aget(url) for url in urls]
    responses = await asyncio.gather(*tasks)
    
    results = []
    for response in responses:
        data = await response.json()
        results.append(data)
    
    return results

# Run async functions
data = asyncio.run(fetch_data())
multiple_data = asyncio.run(fetch_multiple())

Async POST Requests

Asynchronously sends data to the server for processing.

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:
    """
    Asynchronously sends a POST request with optional data payload.
    
    Args:
        url: Target URL
        data: Request body data (supports async iterables)
        json: JSON data to serialize and send
        params: Query parameters
        headers: HTTP headers
        cookies: Cookies
        files: Files for multipart upload
        auth: Authentication
        timeout: Request timeout
        allow_redirects: Follow redirects
        proxies: Proxy configuration
        verify: SSL verification
        stream: Stream response
        cert: Client certificate
        hooks: Request hooks
        retries: Retry configuration
        
    Returns:
        Response object if stream=None/False, AsyncResponse object if stream=True
    """

Usage example:

async def create_user():
    user_data = {
        'name': 'John Doe',
        'email': 'john@example.com',
        'age': 30
    }
    
    response = await niquests.apost(
        'https://api.example.com/users',
        json=user_data
    )
    
    if response.status_code == 201:
        created_user = await response.json()
        return created_user
    else:
        raise Exception(f"Failed to create user: {response.status_code}")

Async PUT Requests

Asynchronously updates or replaces a resource on the server.

async def aput(
    url: str,
    data: 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: AsyncHttpAuthenticationType | 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 | AsyncResponse] | None = None,
    retries: RetryType = DEFAULT_RETRIES,
) -> AsyncResponse:
    """
    Asynchronously sends a PUT request to update or replace a resource.
    """

Async PATCH Requests

Asynchronously performs partial updates on a resource.

async def apatch(
    url: str,
    data: 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: AsyncHttpAuthenticationType | 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 | AsyncResponse] | None = None,
    retries: RetryType = DEFAULT_RETRIES,
) -> AsyncResponse:
    """
    Asynchronously sends a PATCH request for partial resource updates.
    """

Async DELETE Requests

Asynchronously removes a resource from the server.

async def adelete(
    url: str,
    *,
    params: QueryParameterType | None = None,
    headers: HeadersType | None = None,
    cookies: CookiesType | None = None,
    auth: AsyncHttpAuthenticationType | 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 | AsyncResponse] | None = None,
    retries: RetryType = DEFAULT_RETRIES,
    **kwargs: Any,
) -> AsyncResponse:
    """
    Asynchronously sends a DELETE request to remove a resource.
    """

Async HEAD Requests

Asynchronously retrieves only headers without downloading the body content.

async def ahead(
    url: str,
    *,
    params: QueryParameterType | None = None,
    headers: HeadersType | None = None,
    cookies: CookiesType | None = None,
    auth: AsyncHttpAuthenticationType | 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 | AsyncResponse] | None = None,
    retries: RetryType = DEFAULT_RETRIES,
    **kwargs: Any,
) -> AsyncResponse:
    """
    Asynchronously sends a HEAD request to retrieve only response headers.
    
    Note: allow_redirects defaults to False for HEAD requests.
    """

Async OPTIONS Requests

Asynchronously queries server capabilities for a resource.

async def aoptions(
    url: str,
    *,
    params: QueryParameterType | None = None,
    headers: HeadersType | None = None,
    cookies: CookiesType | None = None,
    auth: AsyncHttpAuthenticationType | 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 | AsyncResponse] | None = None,
    retries: RetryType = DEFAULT_RETRIES,
    **kwargs: Any,
) -> AsyncResponse:
    """
    Asynchronously sends an OPTIONS request to query server capabilities.
    """

Async-Specific Features

Concurrent Request Processing

Async functions enable efficient concurrent request processing:

import asyncio
import niquests

async def process_urls_concurrently(urls):
    """Process multiple URLs concurrently for maximum throughput."""
    
    # Create tasks for all requests
    tasks = []
    for url in urls:
        task = niquests.aget(url)
        tasks.append(task)
    
    # Execute all requests concurrently
    responses = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Process results
    results = []
    for i, response in enumerate(responses):
        if isinstance(response, Exception):
            results.append({'url': urls[i], 'error': str(response)})
        else:
            try:
                data = await response.json()
                results.append({'url': urls[i], 'data': data})
            except Exception as e:
                results.append({'url': urls[i], 'error': str(e)})
    
    return results

# Usage
urls = ['https://api.example.com/endpoint1', 'https://api.example.com/endpoint2']
results = asyncio.run(process_urls_concurrently(urls))

Async Body Data

Async functions support async iterables for request body data:

async def stream_data():
    """Example of streaming data asynchronously."""
    
    async def data_generator():
        for i in range(1000):
            yield f"chunk-{i}\n".encode()
            await asyncio.sleep(0.001)  # Simulate async data production
    
    response = await niquests.apost(
        'https://api.example.com/stream',
        data=data_generator(),
        headers={'Content-Type': 'text/plain'}
    )
    
    return await response.text()

Type Definitions

AsyncBodyType = Union[str, bytes, dict, list, IOBase, AsyncIterable[bytes]]
AsyncHttpAuthenticationType = Union[
    Tuple[str, str],
    HTTPBasicAuth,
    BearerTokenAuth,
    Callable[[PreparedRequest], Awaitable[PreparedRequest]]
]

Error Handling

Async functions raise the same exceptions as their synchronous counterparts. Use try-except blocks within async functions:

async def safe_request(url):
    """Example of proper async error handling."""
    try:
        response = await niquests.aget(url, timeout=10.0)
        response.raise_for_status()  # Raise exception for HTTP errors
        return await response.json()
    except niquests.ConnectionError:
        return {'error': 'Connection failed'}
    except niquests.Timeout:
        return {'error': 'Request timed out'}
    except niquests.HTTPError as e:
        return {'error': f'HTTP error: {e.response.status_code}'}
    except Exception as e:
        return {'error': f'Unexpected error: {str(e)}'}

Install with Tessl CLI