Zapros
repository·main·Indexed 41 days ago
https://github.com/kap-sh/zaprosA modern and extensible Python HTTP client requiring Python 3.10 or higher. It features an AsgiHandler for testing ASGI applications (FastAPI, Litestar, Starlette) without network requests, support for browser runtimes via Pyodide, and an optional caching feature using CacheMiddleware and Hishel policies.
What's inside Zapros
- Zapros is a modern and extensible Python HTTP client designed for developers who need a flexible way to handle HTTP requests. It requires Python 3.10 or higher.
Manage Response ownership in handlers
mainWhen a handler receives a
Responsefrom the next handler in the chain, it becomes the owner of that response and is responsible for closing it.To avoid connection leaks, follow these rules:
- Propagate ownership: If you are returning the response (or passing it up the chain), do not close it. The parent handler or client will take responsibility.
- Close and discard: If you decide to return a different response, modify the response, or raise an exception instead of returning the original, you must call
await response.aclose()(async) orresponse.close()(sync) first.
from zapros import AsyncBaseMiddleware, Request, Response class RetryOn503Middleware(AsyncBaseMiddleware): def __init__(self, next_handler: AsyncBaseHandler) -> None: self.async_next = next_handler async def ahandle(self, request: Request) -> Response: response = await self.async_next.ahandle(request) if response.status != 503: return response # propagate ownership await response.aclose() # close because we are discarding it to retry return await self.async_next.ahandle(request)How request matching works in Cassettes
mainCassette interactions are matched based on the method and the normalized URL.
- Query Parameters: These are sorted before matching, so
?a=1&b=2matches?b=2&a=1. - Exclusions: Headers and request bodies are not part of the match key by default.
- Query Parameters: These are sorted before matching, so
How request matching works in Zapros mocks
mainMocks use matchers to determine if an outgoing request should be intercepted. A matcher inspects request components like method, path, host, headers, query, or JSON body. Matchers can be combined usingand_,or_, andnot_, or chained fluently to create complex matching logic.What are Matchers and how do they work?
mainMatchers are objects used to inspect HTTP requests (checking methods, paths, headers, etc.) and return
Trueif the request meets specific criteria. They are primarily used by the mocking system and cassette modifiers.All matchers implement the
Matcherprotocol, which requires amatch(request: Request) -> boolmethod. You can also implement your own custom matchers by following this protocol.from zapros.matchers import Matcher class MyCustomMatcher(Matcher): def match(self, request: Request) -> bool: # logic to return True or False return TrueHow middleware handlers work
mainMiddleware handlers wrap a transport handler to add functionality (like retries, caching, or cookie management). They follow the chain of responsibility pattern and must implement the
AsyncBaseMiddleware(for async clients) orBaseMiddleware(for sync clients) interface.To use middleware, wrap your transport handler with the desired middleware class and pass the resulting object to the
handlerparameter of the client.from zapros import ( AsyncClient, RetryMiddleware, AsyncStdNetworkHandler, ) async with AsyncClient( handler=RetryMiddleware( AsyncStdNetworkHandler(), max_attempts=3, ) ) as client: response = await client.get( "https://api.example.com/data", )How handlers and middleware work in Zapros
mainZapros uses a chain of middleware handlers that wrap each other, ending in a single transport handler that performs the actual I/O (e.g., network calls, ASGI app calls, or browser
fetch()).Every request flows through this chain:
Client ──► Middleware A ──► Middleware B ──► Transport handler ──► network / ASGI app / fetch()Extending Zapros typically involves writing a middleware to intercept requests or responses. The client manages the lifecycle of the chain; calling
client.aclose()or exiting a context manager will automatically walk the chain and callaclose()(orclose()) on every handler and middleware in the sequence.# Conceptual flow Client ──► Middleware A ──► Middleware B ──► Transport handler ──► networkConfigure Caching Policies with SpecificationPolicy and FilterPolicy
mainYou can customize how Zapros decides what to cache by passing a
policyargument toCacheMiddleware. This uses Hishel policies.SpecificationPolicy (Default)
Strictly follows RFC 9111 compliant caching standards.
FilterPolicy
Allows you to cache any response that passes custom filters. You can provide
request_filterandresponse_filterfunctions.Example using FilterPolicy:
from hishel import FilterPolicy from zapros import AsyncClient, CacheMiddleware, AsyncStdNetworkHandler policy = FilterPolicy( request_filter=lambda req: req.method == "GET", response_filter=lambda resp: resp.status == 200, ) client = AsyncClient( handler=CacheMiddleware( AsyncStdNetworkHandler(), policy=policy, ) )from hishel import FilterPolicy from zapros import ( AsyncClient, CacheMiddleware, AsyncStdNetworkHandler, ) policy = FilterPolicy( request_filter=lambda req: req.method == "GET", response_filter=lambda resp: resp.status == 200, ) client = AsyncClient( handler=CacheMiddleware( AsyncStdNetworkHandler(), policy=policy, ), )Use Zapros with Trio
mainIf you use
AsyncStdNetworkHandlerwithin a Trio event loop, it will automatically default to usingTrioTransportinstead ofAsyncIOTransport. You can also explicitly specifyTrioTransportin the handler configuration.from zapros import AsyncClient, AsyncStdNetworkHandler, TrioTransport async with AsyncClient( handler=AsyncStdNetworkHandler(transport=TrioTransport()) ) as client: ...Understand the Sans-IO design of Zapros
mainZapros uses a Sans-IO architecture that separates HTTP semantics from Transport.
- HTTP Semantics: Handled by
AsyncClientandClient. This includes constructing requests, managing headers, decodingContent-Encoding, and encoding body formats (JSON, form, multipart). - Transport: Handled by a separate handler. The handler is responsible for DNS, TCP, TLS, connection pooling, and wire protocols (HTTP/1.1, HTTP/2, HTTP/3).
This separation allows you to use powerful plugins (caching, retries, mocking, cassettes) regardless of the low-level transport implementation, and enables you to swap handlers without changing your client code.
- HTTP Semantics: Handled by
Available Zapros Transport Handlers
mainZapros abstracts the transport layer and provides several specialized handlers to support different environments and performance requirements:
- ASGI Handler: Similar to HTTPX, for interacting with ASGI applications.
- AsyncPyodideHandler: For running the same code in the browser via WASM.
- AsyncPyreqwestHandler: Built on Rust's
reqwestlibrary for high-performance requirements. - WebSocket Client: Built on the
wsprotolibrary. The WebSocket handshake uses your standard client, meaning authentication, cookies, retries, and custom handlers apply to WebSocket connections.
Manage ASGI lifespan with AsgiHandler
mainZapros supports the ASGI lifespan protocol, allowing your application to execute startup and shutdown logic. When
enable_lifespan=Trueis set on theAsgiHandler, the handler manages these lifecycle events.If you are not using the handler as an async context manager, you must manually call
await handler.aclose()to ensure the application shuts down correctly.from litestar import Litestar, get from litestar.datastructures import ( State, ) @get("/state") async def get_state( state: State, ) -> dict: return {"value": state.get("initialized", False)} async def on_startup( app: Litestar, ) -> None: app.state.initialized = True app = Litestar( route_handlers=[get_state], on_startup=[on_startup], ) handler = AsgiHandler(app, enable_lifespan=True) try: async with AsyncClient(handler=handler) as client: response = await client.get( "http://testserver/state", ) print(response.json) finally: await handler.aclose()