runtime

MODULE	DESCRIPTION
annotations
auth
client
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
Response	Declare the expected body and headers for a single HTTP response status code.
Responses	Mapping between response code, headers, media type and body type.
HttpxMiddleware	HTTP middleware protocol, used to fix request or response objects.
ModelBase	A simple base class for request and response models. Pydantic's BaseModel with defaults suitable for typical cases.
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
for_api	Create an :class:APIClient for the given API descriptor class.
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:

.. code-block:: python

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

Referenced by:

•  runtime  Responses  responses

Used by:

•  runtime  Response

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:

.. code-block:: python

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

Response dataclass

Response(body: Body, headers: type[BaseModel] | None = None)

Declare the expected body and headers for a single HTTP response status code.

Used as a value inside the :class:Responses mapping.

Example:

.. code-block:: python

Response(Body({'application/json': Cat}))

Response(Body({'application/json': Cat}), CatListMeta)

Returned by:

•  runtime  Responses  responses

Used by:

•  runtime  Responses

ATTRIBUTE	DESCRIPTION
headers	Pydantic model to deserialize response headers into. Fields must be annotated with :class:Header or :class:StatusCode. TYPE: type[BaseModel] | None

headers class-attribute instance-attribute

headers: type[BaseModel] | None = None

Pydantic model to deserialize response headers into. Fields must be annotated with :class:Header or :class:StatusCode.

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

HttpxMiddleware

Bases: Protocol

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

Used by:

•  runtime  for_api

ModelBase

ModelBase(**data: Any)

Bases: BaseModel

A simple base class for request and response models. Pydantic's BaseModel with defaults suitable for typical cases.

Source code in pydantic/main.py

def __init__(self, /, **data: Any) -> None:
    """Create a new model by parsing and validating input data from keyword arguments.

    Raises [`ValidationError`][pydantic_core.ValidationError] if the input data cannot be
    validated to form a valid model.

    `self` is explicitly positional-only to allow `self` as a field name.
    """
    # `__tracebackhide__` tells pytest and some other tools to omit this function from tracebacks
    __tracebackhide__ = True
    validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
    if self is not validated_self:
        warnings.warn(
            'A custom validator is returning a value other than `self`.\n'
            "Returning anything other than `self` from a top level model validator isn't supported when validating via `__init__`.\n"
            'See the `model_validator` docs (https://docs.pydantic.dev/latest/concepts/validators/#model-validators) for more details.',
            stacklevel=2,
        )

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)

for_api

for_api(api: type[API_T], client: AsyncClient, base_url: str, *, middlewares: Sequence[HttpxMiddleware] = (), auth: Auth | None = None) -> APIClient[API_T]

Create an :class:APIClient for the given API descriptor class.

PARAMETER	DESCRIPTION
api	The API descriptor class — a plain class whose methods are decorated with :func:get, :func:post, etc. TYPE: type[API_T]
client	The underlying :class:httpx.AsyncClient used to send requests. TYPE: AsyncClient
base_url	Base URL prepended to all operation paths. TYPE: str
middlewares	Optional sequence of :class:HttpxMiddleware instances applied to every request. TYPE: Sequence[HttpxMiddleware] DEFAULT: ()
auth	Optional :class:httpx.Auth instance used to authenticate requests. TYPE: Auth | None DEFAULT: None

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

def for_api(
    api: type[API_T],
    client: httpx.AsyncClient,
    base_url: str,
    *,
    middlewares: Sequence[HttpxMiddleware] = (),
    auth: httpx.Auth | None = None,
) -> APIClient[API_T]:
    """
    Create an :class:`APIClient` for the given API descriptor class.

    :param api: The API descriptor class — a plain class whose methods are decorated with :func:`get`, :func:`post`, etc.
    :param client: The underlying :class:`httpx.AsyncClient` used to send requests.
    :param base_url: Base URL prepended to all operation paths.
    :param middlewares: Optional sequence of :class:`HttpxMiddleware` instances applied to every request.
    :param auth: Optional :class:`httpx.Auth` instance used to authenticate requests.
    """
    api_model = APIModel(api)
    return APIClient(api_model, client, base_url, auth=auth, middlewares=middlewares)

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.ops.fn, 'cursor', get_cursor_from_response)(parameter=value): # Process page

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::

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

    :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