loamy 0.0.2a0

No project description provided

Overview

This project allows you to execute a list of http operations asynchronously from within an synchronous context.

It does not care whether you should do this. It simply allows you to do so if you desire.

Installing

The package is available via pip.

pip install loamy

If you're not on Windows, install the uvloop extra to increase performance.

pip install "loamy[uvloop]"

Usage

The package can be imported as shown:

from loamy.session import Clump, RequestMap, RequestResults

Class	Description
Clump	Container object that stores collection of requests (type RequestMap) to send
RequestMap	Container object that stores all info about an individual request to send
RequestResults	Container object that stores the request responses and any exceptions raised

Example

# Create RequestMap objects
req1 = RequestMap(
    url="https://baconipsum.com/api",
    httpOperation="GET",
    queryParams={"type": "meat-and-filler", "format": "json"},
)
req2 = RequestMap(
    url="https://baconipsum.com/api",
    httpOperation="GET",
    queryParams={"type": "all-meat", "format": "json"},
)
req3 = RequestMap(
    url="https://baconipsum.com/api",
    httpOperation="GET",
    queryParams={"type": "meat-and-filler", "format": "json"},
)

# Create Clump and call sendRequests()
session = Clump(requests=[req1, req2, req3])
reqResps: RequestResults = session.sendRequests(return_exceptions=True)

# Handle exceptions raised for individual requests
if len(reqResps.taskExceptions) > 0:
    print("Handling exceptions")

# Handle responses for individual requests
for resp in requestResponses:
    httpVerb = resp.requestMap.httpOperation
    print(f"Evaluating response for {httpVerb} request to {resp.requestMap.url}")
    print(f"Status Code: {resp.statusCode}")
    if resp.body is not None:
        print(resp.body)

RequestMap Class

class RequestMap(msgspec.Struct):
    url: str
    httpOperation: Literal["GET", "POST", "PUT", "PATCH", "OPTIONS", "DELETE"]
    body: dict | None = None
    queryParams: dict[str, str] | None = None
    headers: dict[str, str] | None = None

RequestResponse Class

class RequestResponse(msgspec.Struct):
    requestMap: RequestMap
    statusCode: int
    body: dict | None = None

RequestResults Class

@dataclass
class RequestResults:
    requestResponses: list[RequestResponse]
    taskExceptions: list[BaseException]