runtime

MODULE	DESCRIPTION
annotations
auth
client_base
middleware
model
paging
types_

CLASS	DESCRIPTION
Body	Link content type headers with a python type.
Cookie
Header	Mark parameter as HTTP Header
Metadata	Annotation for models that hold other WebArg fields.
Path
Query
Responses	Mapping between response code, headers, media type and body type.
ClientBase	Base for Client classes
HttpxMiddleware	HTTP middleware protocol, used to fix request or response objects.
HttpErrorResponse	Base error class for declared HTTP error responses - 4XX & 5XX.
LapidaryResponseError	Base class for errors that wrap the response
UnexpectedResponse	Raised when the remote server responded with code and content-type pair that wasn't declared in the method return annotation
FormExplode

FUNCTION	DESCRIPTION
iter_pages	Take a function that returns a pageg response and return a function that returns an async iterator that iterates over the pages.

Body dataclass

Body(content: Mapping[MimeType, type])

Bases: WebArg

Link content type headers with a python type. When used with a method parameter, it tells lapidary what content-type header to send for a given body type. When used in return annotation, it tells lapidary the type to process the response body as.

Example use with parameter:

body: Body({'application/json': BodyModel})

Referenced by:

•  runtime  Responses  responses

Cookie

Cookie(alias: str | None = None, /, *, style: type[MultimapSerializationStyle] = FormExplode)

Bases: Param

PARAMETER	DESCRIPTION
alias	Cookie name, if different than the name of the annotated parameter TYPE: str | None DEFAULT: None
style	Serialization style TYPE: type[MultimapSerializationStyle] DEFAULT: FormExplode

Source code in src/lapidary/runtime/annotations.py

def __init__(
    self,
    alias: str | None = None,
    /,
    *,
    style: type[MultimapSerializationStyle] = FormExplode,
) -> None:
    """
    :param alias: Cookie name, if different than the name of the annotated parameter
    :param style: Serialization style
    """
    super().__init__(alias)
    self.style = style

Header

Header(alias: str | None = None, /, *, style: type[MultimapSerializationStyle] = SimpleMultimap)

Bases: Param

Mark parameter as HTTP Header

PARAMETER	DESCRIPTION
alias	Header name, if different than the name of the annotated parameter TYPE: str | None DEFAULT: None
style	Serialization style TYPE: type[MultimapSerializationStyle] DEFAULT: SimpleMultimap

Source code in src/lapidary/runtime/annotations.py

def __init__(
    self,
    alias: str | None = None,
    /,
    *,
    style: type[MultimapSerializationStyle] = SimpleMultimap,
) -> None:
    """
    :param alias: Header name, if different than the name of the annotated parameter
    :param style: Serialization style
    """
    super().__init__(alias)
    self.style = style

Metadata

Bases: WebArg

Annotation for models that hold other WebArg fields. Can be used to group request parameters as an alternative to passing parameters directly.

Example:

class RequestMetadata(pydantic.BaseModel):
    my_header: typing.Annotated[
        str,
        Header('my-header'),
    ]

class Client(ApiClient):
@get(...)
async def my_method(
    headers: Annotated[RequestMetadata, Metadata]
):

Path

Path(alias: str | None = None, /, *, style: type[StringSerializationStyle] = SimpleString)

Bases: Param

PARAMETER	DESCRIPTION
alias	Path parameter name, if different than the name of the annotated parameter TYPE: str | None DEFAULT: None
style	Serialization style TYPE: type[StringSerializationStyle] DEFAULT: SimpleString

Source code in src/lapidary/runtime/annotations.py

def __init__(
    self,
    alias: str | None = None,
    /,
    *,
    style: type[StringSerializationStyle] = SimpleString,
) -> None:
    """
    :param alias: Path parameter name, if different than the name of the annotated parameter
    :param style: Serialization style
    """
    super().__init__(alias)
    self.style = style

Query

Query(alias: str | None = None, /, *, style: type[MultimapSerializationStyle] = FormExplode)

Bases: Param

PARAMETER	DESCRIPTION
alias	Query parameter name, if different than the name of the annotated parameter TYPE: str | None DEFAULT: None
style	Serialization style TYPE: type[MultimapSerializationStyle] DEFAULT: FormExplode

Source code in src/lapidary/runtime/annotations.py

def __init__(
    self,
    alias: str | None = None,
    /,
    *,
    style: type[MultimapSerializationStyle] = FormExplode,
) -> None:
    """
    :param alias: Query parameter name, if different than the name of the annotated parameter
    :param style: Serialization style
    """
    super().__init__(alias)
    self.style = style

Responses dataclass

Responses(responses: Mapping[StatusCodeRange, Response])

Bases: WebArg

Mapping between response code, headers, media type and body type. The simplified structure is:

response code => (
    body: content type => body model type
    headers model
)

The structure follows OpenAPI 3.

ATTRIBUTE	DESCRIPTION
responses	Map of status code match to Response. TYPE: Mapping[StatusCodeRange, Response]

responses instance-attribute

responses: Mapping[StatusCodeRange, Response]

Map of status code match to Response. Key may be:

• any HTTP status code
• HTTP status code range, i.e. 1XX, 2XX, etc
• "default"

The most specific value takes precedence.

Value is Body

ClientBase

ClientBase(client: AsyncClient | None = None, base_url: str | None = None, *, middlewares: Sequence[HttpxMiddleware] = (), auth: Auth | None = None)

Bases: ABC

Base for Client classes

PARAMETER	DESCRIPTION
client	the httpx client to use TYPE: AsyncClient | None DEFAULT: None
middlewares	list of middlewares to process HTTP requests and responses TYPE: Sequence[HttpxMiddleware] DEFAULT: ()

Source code in src/lapidary/runtime/client_base.py

def __init__(
    self,
    client: httpx.AsyncClient | None = None,
    base_url: str | None = None,
    *,
    middlewares: Sequence[HttpxMiddleware] = (),
    auth: httpx.Auth | None = None,
) -> None:
    """
    :param client: the httpx client to use
    :param middlewares: list of middlewares to process HTTP requests and responses
    """
    self._base_url = base_url
    self._client = client or httpx.AsyncClient()
    self._auth = auth
    self._middlewares = middlewares
    self._send = _mk_send(self._client.send, self._auth, self._middlewares)

HttpxMiddleware

Bases: Protocol

HTTP middleware protocol, used to fix request or response objects.

Used by:

•  runtime  ClientBase

HttpErrorResponse

HttpErrorResponse(status_code: int, headers: Headers, body: Body)

Bases: Generic[Body, Headers], LapidaryResponseError

Base error class for declared HTTP error responses - 4XX & 5XX. Python doesn't fully support parametrized exception types, but extending types can concretize it.

Source code in src/lapidary/runtime/model/error.py

def __init__(self, status_code: int, headers: Headers, body: Body):
    super().__init__()
    self.status_code = status_code
    self.headers = headers
    self.body = body

LapidaryResponseError

Bases: LapidaryError

Base class for errors that wrap the response

Subclassed by:

•  runtime
  •  HttpErrorResponse
  •  UnexpectedResponse

UnexpectedResponse

UnexpectedResponse(response: Response)

Bases: LapidaryResponseError

Raised when the remote server responded with code and content-type pair that wasn't declared in the method return annotation

Source code in src/lapidary/runtime/model/error.py

def __init__(self, response: httpx.Response):
    self.response = response
    self.content_type = response.headers.get('content-type')

FormExplode

Bases: MultimapSerializationStyle

Used by:

•  runtime
  •  Cookie
  •  Query

METHOD	DESCRIPTION
serialize_object	Disregard name, return a map of {key: value}

serialize_object classmethod

serialize_object(_name: str, value: ObjectType) -> Multimap

Disregard name, return a map of {key: value}

Source code in src/lapidary/runtime/model/param_serialization.py

@classmethod
def serialize_object(cls, _name: str, value: ObjectType) -> Multimap:
    """Disregard name, return a map of {key: value}"""
    return itertools.chain.from_iterable(cls.serialize_scalar(key, item) for key, item in value.items() if item)

iter_pages

iter_pages(fn: Callable[P, Awaitable[R]], cursor_param_name: str, get_cursor: Callable[[R], C | None]) -> Callable[P, AsyncIterable[R]]

Take a function that returns a pageg response and return a function that returns an async iterator that iterates over the pages.

The returned function can be called with the same parameters as :param:fn (except for the cursor parameter), and returns an async iterator that yields results from :param:fn, handling pagination automatically.

The function :param:fn will be called initially without the cursor parameter and then called with the cursor parameter as long as :param:get_cursor can extract a cursor from the result.

Example:

async for page in iter_pages(client.fn, 'cursor', extractor_fn)(parameter=value):
    # Process page

Typically, an API will use the same paging pattern for all operations supporting it, so it's a good idea to write a shortcut function:

from lapidary.runtime import iter_pages as _iter_pages

def iter_pages[P, R](fn: Callable[P, Awaitable[R]]) -> Callable[P, AsyncIterable[R]]:
    return _iter_pages(fn, 'cursor', lambda result: ...)

PARAMETER	DESCRIPTION
fn	An async function that retrieves a page of data. TYPE: Callable[P, Awaitable[R]]
cursor_param_name	The name of the cursor parameter in :param:fn. TYPE: str
get_cursor	A function that extracts a cursor value from the result of :param:fn. Return None to end the iteration. TYPE: Callable[[R], C | None]

Source code in src/lapidary/runtime/paging.py

def iter_pages(
    fn: Callable[P, Awaitable[R]],
    cursor_param_name: str,
    get_cursor: Callable[[R], C | None],
) -> Callable[P, AsyncIterable[R]]:
    """
    Take a function that returns a pageg response and return a function that returns an async iterator that iterates over the pages.

    The returned function can be called with the same parameters as :param:`fn` (except for the cursor parameter),
    and returns an async iterator that yields results from :param:`fn`, handling pagination automatically.

    The function :param:`fn` will be called initially without the cursor parameter and then called with the cursor parameter
    as long as :param:`get_cursor` can extract a cursor from the result.

    **Example:**

    ```python
    async for page in iter_pages(client.fn, 'cursor', extractor_fn)(parameter=value):
        # Process page
    ```

    Typically, an API will use the same paging pattern for all operations supporting it, so it's a good idea to write a shortcut function:

    ```python
    from lapidary.runtime import iter_pages as _iter_pages

    def iter_pages[P, R](fn: Callable[P, Awaitable[R]]) -> Callable[P, AsyncIterable[R]]:
        return _iter_pages(fn, 'cursor', lambda result: ...)
    ```

    :param fn: An async function that retrieves a page of data.
    :param cursor_param_name: The name of the cursor parameter in :param:`fn`.
    :param get_cursor: A function that extracts a cursor value from the result of :param:`fn`. Return `None` to end the iteration.
    """

    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> AsyncIterable[R]:
        result = await fn(*args, **kwargs)  # type: ignore[call-arg]
        yield result
        cursor = get_cursor(result)

        while cursor:
            kwargs[cursor_param_name] = cursor
            result = await fn(*args, **kwargs)  # type: ignore[call-arg]
            yield result

            cursor = get_cursor(result)

    return wrapper