Zapros

repository·main·Indexed 41 days ago

https://github.com/kap-sh/zapros

A 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.

Tokens
46.8K
Snippets
157
Records
179
Agent score
52%

What's inside Zapros

  1. Manage Response ownership in handlers

    main

    When a handler receives a Response from 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:

    1. 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.
    2. 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) or response.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)
  2. How request matching works in Cassettes

    main

    Cassette interactions are matched based on the method and the normalized URL.

    • Query Parameters: These are sorted before matching, so ?a=1&b=2 matches ?b=2&a=1.
    • Exclusions: Headers and request bodies are not part of the match key by default.
  3. How request matching works in Zapros mocks

    main
    Mocks 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 using and_, or_, and not_, or chained fluently to create complex matching logic.
  4. What are Matchers and how do they work?

    main

    Matchers are objects used to inspect HTTP requests (checking methods, paths, headers, etc.) and return True if the request meets specific criteria. They are primarily used by the mocking system and cassette modifiers.

    All matchers implement the Matcher protocol, which requires a match(request: Request) -> bool method. 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 True
  5. How middleware handlers work

    main

    Middleware 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) or BaseMiddleware (for sync clients) interface.

    To use middleware, wrap your transport handler with the desired middleware class and pass the resulting object to the handler parameter 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",
        )
  6. How handlers and middleware work in Zapros

    main

    Zapros 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 call aclose() (or close()) on every handler and middleware in the sequence.

    # Conceptual flow
    Client ──► Middleware A ──► Middleware B ──► Transport handler ──► network
  7. Configure Caching Policies with SpecificationPolicy and FilterPolicy

    main

    You can customize how Zapros decides what to cache by passing a policy argument to CacheMiddleware. 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_filter and response_filter functions.

    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,
        ),
    )
  8. Use Zapros with Trio

    main

    If you use AsyncStdNetworkHandler within a Trio event loop, it will automatically default to using TrioTransport instead of AsyncIOTransport. You can also explicitly specify TrioTransport in the handler configuration.

    from zapros import AsyncClient, AsyncStdNetworkHandler, TrioTransport
    
    async with AsyncClient(
        handler=AsyncStdNetworkHandler(transport=TrioTransport())
    ) as client:
        ...
  9. Understand the Sans-IO design of Zapros

    main

    Zapros uses a Sans-IO architecture that separates HTTP semantics from Transport.

    • HTTP Semantics: Handled by AsyncClient and Client. This includes constructing requests, managing headers, decoding Content-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.

  10. Available Zapros Transport Handlers

    main

    Zapros 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 reqwest library for high-performance requirements.
    • WebSocket Client: Built on the wsproto library. The WebSocket handshake uses your standard client, meaning authentication, cookies, retries, and custom handlers apply to WebSocket connections.
  11. Manage ASGI lifespan with AsgiHandler

    main

    Zapros supports the ASGI lifespan protocol, allowing your application to execute startup and shutdown logic. When enable_lifespan=True is set on the AsgiHandler, 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()