Lapidary Reference

ClientBase

Bases: ABC

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

class ClientBase(abc.ABC):
    def __init__(
        self,
        security: Iterable[SecurityRequirements] | None = None,
        session_factory: SessionFactory = httpx.AsyncClient,
        middlewares: Sequence[HttpxMiddleware] = (),
        **httpx_kwargs: typing.Unpack[ClientArgs],
    ) -> None:
        self._client = session_factory(**httpx_kwargs)
        if USER_AGENT not in self._client.headers:
            self._client.headers[USER_AGENT] = lapidary_user_agent()

        self._auth_registry = AuthRegistry(security)
        self._middlewares = middlewares

    async def __aenter__(self: typing.Self) -> typing.Self:
        await self._client.__aenter__()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None = None,
        exc_value: BaseException | None = None,
        traceback: types.TracebackType | None = None,
    ) -> bool | None:
        return await self._client.__aexit__(exc_type, exc_value, traceback)

    def lapidary_authenticate(self, *auth_args: NamedAuth, **auth_kwargs: httpx.Auth) -> None:
        """Register named Auth instances for future use with methods that require authentication."""
        if auth_args:
            # make python complain about duplicate names
            self.lapidary_authenticate(**dict(auth_args), **auth_kwargs)

        self._auth_registry.authenticate(auth_kwargs)

    def lapidary_deauthenticate(self, *sec_names: str) -> None:
        """Remove reference to a given Auth instance.
        Calling with no parameters removes all references"""

        self._auth_registry.deauthenticate(sec_names)

lapidary_authenticate(*auth_args, **auth_kwargs)

Register named Auth instances for future use with methods that require authentication.

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

def lapidary_authenticate(self, *auth_args: NamedAuth, **auth_kwargs: httpx.Auth) -> None:
    """Register named Auth instances for future use with methods that require authentication."""
    if auth_args:
        # make python complain about duplicate names
        self.lapidary_authenticate(**dict(auth_args), **auth_kwargs)

    self._auth_registry.authenticate(auth_kwargs)

lapidary_deauthenticate(*sec_names)

Remove reference to a given Auth instance. Calling with no parameters removes all references

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

def lapidary_deauthenticate(self, *sec_names: str) -> None:
    """Remove reference to a given Auth instance.
    Calling with no parameters removes all references"""

    self._auth_registry.deauthenticate(sec_names)

FormExplode

Bases: MultimapSerializationStyle

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

class FormExplode(MultimapSerializationStyle):
    @classmethod
    def serialize_scalar(cls, name: str, value: ScalarType) -> Multimap:
        return SimpleMultimap.serialize_scalar(name, value)

    @classmethod
    def serialize_array(cls, name: str, value: ArrayType) -> Multimap:
        return itertools.chain.from_iterable(cls.serialize_scalar(name, item) for item in value)

    @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)

    @classmethod
    def deserialize_array(cls, value: str, _) -> ArrayType:
        raise NotImplementedError

serialize_object(_name, value) classmethod

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

Source code in docs/lapidary/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)

HttpErrorResponse

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 docs/lapidary/src/lapidary/runtime/model/error.py

class HttpErrorResponse(typing.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.
    """

    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

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

class LapidaryResponseError(LapidaryError):
    """Base class for errors that wrap the response"""

Metadata

Bases: WebArg

Annotation for models that hold other WebArg fields

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

class Metadata(WebArg):
    """Annotation for models that hold other WebArg fields"""

UnexpectedResponse

Bases: LapidaryResponseError

Base error class for undeclared responses

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

class UnexpectedResponse(LapidaryResponseError):
    """Base error class for undeclared responses"""

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

iter_pages(fn, cursor_param_name, get_cursor)

Create a function that returns an async iterator over pages from the async operation function :param:fn.

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:

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

.. code:: 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: ...)

Parameters:

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

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

def iter_pages(
    fn: Callable[P, Awaitable[R]],
    cursor_param_name: str,
    get_cursor: Callable[[R], Optional[C]],
) -> Callable[P, AsyncIterable[R]]:
    """
    Create a function that returns an async iterator over pages from the async operation function :param:`fn`.

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

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

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