req

This module is designed to handle common HTTP operations in a clean and reusable manner, with dynamic endpoint resolution and robust response management.

Let’s run a micro server:

# test.py
from usrv import route

# allow req.Endpoint.connect
@route.bind("/", methods=["HEAD"])
def base():
    return 200,

# public key endoint for encryption
@route.bind("/puk", methods=["GET"])
def puk():
    return 200, route.PUBLIC_KEY

@route.bind("/index")
def index(*args):
    return (200, ) + args

@route.bind("/api/endpoint", methods=["GET", "POST"])
def endpoit(a, b, **kwargs):
    method = kwargs["method"]
    if method == "POST":
        return 202, kwargs["data"]
    elif method == "GET":
        return 200, a, b
    else:
        return 404,

route.run(host='127.0.0.1', port=5000)

$ python path/to/test.py
INFO:usrv:listening on http://127.0.0.1:5000
CTRL+C to stop...

Connect to a peer

Remote path / allows HEAD request:

>>> from usrv import req
>>> req.Endpoint.connect("http://127.0.0.1:5000")
200

Else, maunally set peer to req.Endpoint or use _peer keyword on each request :

>>> from usrv import req
>>> req.ENDPOINT.peer = "http://127.0.0.1:5000"
>>> # or
>>> req.GET.api.endpoint(_peer="http://127.0.0.1:5000")
>>> req.POST.api.endpoint(_peer="http://127.0.0.1:5000")
>>> # ...

Endpoints

>>> # GET http://127.0.0.1:5000/puk
>>> req.GET.puk()
pP15aGDcFoqGTHTReiIfEvUcQ2c3AQjYcgCeLgKhpa38Rsub69i6RifuYPGtOOyld7j6y0LP6i0aqBuFYcSmTQ==
>>> # GET http://127.0.0.1:5000/api/endpoints?a=12&b=test
>>> req.GET.api.endpoint(a=12, b="test")
["12", "test"]
>>> # POST data to http://127.0.0.1:5000/api/endpoints
>>> req.POST.api.endpoint(value1=1, value2=2)
'{"value1": 1, "value2": 2}'

Encrypt HTTP body

>>> # encrypt only server body response
>>> req.POST.api.endpoint(
...   value1=1, value2=2, _headers={"Sender-Public-Key:req.PUBLIC_KEY}
... )
'{"value1": 1, "value2": 2}'
>>> # encrypt request and response bodies
>>> puk = req.GET.puk()
>>> puk
pP15aGDcFoqGTHTReiIfEvUcQ2c3AQjYcgCeLgKhpa38Rsub69i6RifuYPGtOOyld7j6y0LP6i0aqBuFYcSmTQ==
>>> req.POST.api.endpoint(value1=1, value2=2, _puk=puk)
'{"value1": 1, "value2": 2}'

>>> from usrv import secp256k1
>>> # generate a random keypair
>>> prk, puk = secp256k1.generate_keypair()
>>> # target public key is not server public key
>>> puk == req.GET.puk()
False
>>> print(req.POST.api.endpoints(value1=1, value2=2, _puk=puk))
<!DOCTYPE HTML>
<html lang="en">
    <head>
        <meta charset="utf-8">
        <title>Error response</title>
    </head>
    <body>
    <h1>Error response</h1>
        <p>Error code: 500</p>
        <p>Message: Encryption error.</p>
        <p>Error code explanation: 5e234e6f68f30a056c2bd53e97b785e49895b1ae85d3cf323a95ot encrypted for public key pP15aGDcFoqGTHTReiIfEvUcQ2c3AQjYcgCeLgKhpa38Rsub69i6RifuYPGtOOyld7j6y0LP6i0aqBuFYcSmTQ==.</p>
    </body>
</html>
>>> # target public key is not the client public key
>>> req.POST.api.endpoint(
...   value1=1, value2=2, _headers={"Sender-Public-Key":puk}
... )
ERROR:usrv:Encryption error:
cc72124506d28ccb7bda70d6649f3f007bca1b8f1d829b047267ea543aa34c96ab39 not encrypted for public key pP15aGDcFoqGTHTReiIfEvUcQ2c3AQjYcgCeLgKhpa38Rsub69i6RifuYPGtOOyld7j6y0LP6i0aqBuFYcSmTQ==
'cc72124506d28ccb7bda70d6649f3f007bca1b8f1d829b047267ea543aa34c96ab39'
>>>

RequestCache Objects

class RequestCache()

Cache manager for HTTP Request objects.

RequestCache.init

def __init__(max_size: int = 100, ttl: int = 300)

Initialize the cache.

Arguments:

max_size int - Maximum number of entries in the cache.
ttl int - Time-to-live for cache entries in seconds.

RequestCache.generate_key

@staticmethod
def generate_key(method: str, path: str, **kwargs) -> str

Generate a unique cache key for a request.

RequestCache.get

def get(key: str) -> typing.Union[None, typing.Tuple[int, Request]]

Retrieve an entry from the cache.

RequestCache.set

def set(key: str, request: Request) -> None

Add an entry to the cache.

FormData Objects

class FormData(list)

Implementation of multipart/form-data encoder.

This class provides methods to construct, encode, and decode multipart/form-data content, as described in RFC 7578.

FormData.append_json

def append_json(name: str, value: dict = {}, **kwval) -> None

Add a JSON object to the multipart body.

Arguments:

name str - The name of the form field.
value dict, optional - A dictionary representing the JSON object. Defaults to None.
kwval - Additional key-value pairs to include in the JSON object.

Returns:

typing.Any - The updated FormData instance.

FormData.append_value

def append_value(name: str, value: typing.Union[str, bytes],
                 **headers) -> None

Add a text or binary value to the multipart body.

Arguments:

name str - The name of the form field.
value Union[str, bytes] - The value to add. Can be a string or bytes.
headers - Additional headers to include for this field.

FormData.append_file

def append_file(name: str, path: str) -> typing.Any

Add a file to the multipart body.

Arguments:

name str - The name of the form field.
path str - The path to the file to be added.

Raises:

IOError - If the file does not exist.

FormData.dumps

def dumps() -> str

Encode the FormData instance as a multipart/form-data body.

Returns:

str - The encoded body and the corresponding Content-Type header.

FormData.dump

def dump(folder: str = None) -> None

Save the FormData instance to files in a directory.

Each field in the FormData is written to a separate file. Additional metadata is saved as JSON.

Returns:

None

FormData.encode

@staticmethod
def encode(data: dict) -> str

Encode a dictionary as a multipart/form-data string.

Arguments:

data dict - The data to encode. Can include filepath, strings, or FormData instances.

Returns:

str - The encoded multipart/form-data string.

FormData.decode

@staticmethod
def decode(data: str) -> typing.Any

Decode a multipart/form-data string into a FormData instance.

Arguments:

data str - The multipart/form-data string to decode.

Returns:

FormData - The decoded FormData instance.

Endpoint Objects

class Endpoint()

Represents an HTTP endpoint with dynamic attribute handling.

Attributes:

startswith_ re.Pattern - Pattern to match internal attributes.
timeout int - Default timeout for requests.
opener OpenerDirector - Opener to handle HTTP requests.
peer str - Base URL for the endpoint.

Endpoint.init

def __init__(master: typing.Any = None,
             name: str = "",
             method: Callable = None) -> None

Initializes an Endpoint instance.

Arguments:

master typing.Any - Parent endpoint.
name str - Name of the current endpoint.
method Callable - Request-building method.

Endpoint.getattr

def __getattr__(attr: str) -> typing.Any

Dynamically resolves sub-endpoints.

Arguments:

attr str - Attribute name.

Returns:

Endpoint - New sub-endpoint instance.

Endpoint.call

def __call__(*args, **kwargs) -> typing.Any

Executes the endpoint’s method with provided arguments.

Arguments:

**kwargs - Parameters for the HTTP request.

Returns:

typing.Any - value returned by method attribute.

Endpoint.connect

@staticmethod
def connect(peer: str) -> typing.Union[int, bool]

Tests connection to a peer endpoint and store it if success.

Arguments:

peer str - Peer URL to test.

Returns:

typing.Union[int, bool]: HTTP status code or False on failure.

build_request

def build_request(method: str = "GET", path: str = "/", **kwargs) -> Request

Builds an HTTP request object.

Arguments:

method str - HTTP method (e.g., ‘GET’, ‘POST’). Defaults to ‘GET’.
path str - URL path for the request. Defaults to ‘/’.
**kwargs - Additional keyword arguments for query parameters, headers, and data.

Returns:

Request - Configured HTTP request object.

manage_response

def manage_response(resp: HTTPResponse) -> typing.Union[dict, str]

Parses the HTTP response.

Arguments:

resp HTTPResponse - HTTP response object.

Returns:

typing.Union[dict, str]: Decoded response content.

build_endpoint

def build_endpoint(http_req: str = "GET",
                   encoder: Callable = json.dumps,
                   timeout: int = Endpoint.timeout) -> Endpoint

Creates a root endpoint.

Arguments:

http_req str - Name of HTTP method (i.e. HEAD, GET, POST etc…).
encoder Callable - Data encoder function to use. (defaults to json)
timeout int - Request timeout in seconds.

Returns:

Endpoint - Root endpoint.

Keys	Action
`?`	Open this help
`n`	Next page
`p`	Previous page
`s`	Search

req

RequestCache Objects

RequestCache.__init__

RequestCache.generate_key

RequestCache.get

RequestCache.set

FormData Objects

FormData.append_json

FormData.append_value

FormData.append_file

FormData.dumps

FormData.dump

FormData.encode

FormData.decode

Endpoint Objects

Endpoint.__init__

Endpoint.__getattr__

Endpoint.__call__

Endpoint.connect

build_request

manage_response

build_endpoint

RequestCache.init

Endpoint.init

Endpoint.getattr

Endpoint.call