Lapidary Reference

Body dataclass

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

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

@dc.dataclass
class Body(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:

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

    content: Mapping[MimeType, type]

ClientBase

Bases: ABC

Base for Client classes

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

class ClientBase(abc.ABC):
    """Base for Client classes"""

    def __init__(
        self,
        security: Iterable[SecurityRequirements] | None = None,
        session_factory: SessionFactory = httpx.AsyncClient,
        middlewares: Sequence[HttpxMiddleware] = (),
        **httpx_kwargs: typing.Unpack[ClientArgs],
    ) -> None:
        """
        :param security: Security requirements as a mapping of name => list of scopes
        :param session_factory: `httpx.AsyncClient` or a subclass type
        :param middlewares: list of middlewares to process HTTP requests and responses
        :param httpx_kwargs: keyword arguments to pass to session_factory
        """
        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.
        Accepts named [`Auth`][httpx.Auth] as tuples name, auth or as named arguments
        """
        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)

__init__(security=None, session_factory=httpx.AsyncClient, middlewares=(), **httpx_kwargs)

Parameters:

Name	Type	Description	Default
security	Iterable[SecurityRequirements] | None	Security requirements as a mapping of name => list of scopes	None
session_factory	SessionFactory	httpx.AsyncClient or a subclass type	AsyncClient
middlewares	Sequence[HttpxMiddleware]	list of middlewares to process HTTP requests and responses	()
httpx_kwargs	Unpack[ClientArgs]	keyword arguments to pass to session_factory	{}

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

def __init__(
    self,
    security: Iterable[SecurityRequirements] | None = None,
    session_factory: SessionFactory = httpx.AsyncClient,
    middlewares: Sequence[HttpxMiddleware] = (),
    **httpx_kwargs: typing.Unpack[ClientArgs],
) -> None:
    """
    :param security: Security requirements as a mapping of name => list of scopes
    :param session_factory: `httpx.AsyncClient` or a subclass type
    :param middlewares: list of middlewares to process HTTP requests and responses
    :param httpx_kwargs: keyword arguments to pass to session_factory
    """
    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

lapidary_authenticate(*auth_args, **auth_kwargs)

Register named Auth instances for future use with methods that require authentication. Accepts named [Auth][httpx.Auth] as tuples name, auth or as named arguments

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.
    Accepts named [`Auth`][httpx.Auth] as tuples name, auth or as named arguments
    """
    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)

Cookie

Bases: Param

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

class Cookie(Param):
    def __init__(
        self,
        alias: typing.Optional[str] = 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

__init__(alias=None, /, *, style=FormExplode)

Parameters:

Name	Type	Description	Default
alias	Optional[str]	Cookie name, if different than the name of the annotated parameter	None
style	type[MultimapSerializationStyle]	Serialization style	FormExplode

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

def __init__(
    self,
    alias: typing.Optional[str] = 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

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)

Header

Bases: Param

Mark parameter as HTTP Header

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

class Header(Param):
    """Mark parameter as HTTP Header"""

    def __init__(
        self,
        alias: typing.Optional[str] = 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

__init__(alias=None, /, *, style=SimpleMultimap)

Parameters:

Name	Type	Description	Default
alias	Optional[str]	Header name, if different than the name of the annotated parameter	None
style	type[MultimapSerializationStyle]	Serialization style	SimpleMultimap

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

def __init__(
    self,
    alias: typing.Optional[str] = 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

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

HttpxMiddleware

Bases: Generic[State]

Base class for HTTP middleware.

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

class HttpxMiddleware(Generic[State]):
    """
    Base class for HTTP middleware.
    """

    @abc.abstractmethod
    async def handle_request(self, request: httpx.Request) -> State:
        """Called for each request after it's generated for a method call but before it's sent to the remote server.
        Any returned value will be passed back to handle_response.
        """

    async def handle_response(self, response: httpx.Response, request: httpx.Request, state: State) -> None:
        """Called for each response after it's been received from the remote server and before it's converted to the return type as defined
        in the python method.

        state is the value returned by handle_request
        """

handle_request(request) abstractmethod async

Called for each request after it's generated for a method call but before it's sent to the remote server. Any returned value will be passed back to handle_response.

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

@abc.abstractmethod
async def handle_request(self, request: httpx.Request) -> State:
    """Called for each request after it's generated for a method call but before it's sent to the remote server.
    Any returned value will be passed back to handle_response.
    """

handle_response(response, request, state) async

Called for each response after it's been received from the remote server and before it's converted to the return type as defined in the python method.

state is the value returned by handle_request

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

async def handle_response(self, response: httpx.Response, request: httpx.Request, state: State) -> None:
    """Called for each response after it's been received from the remote server and before it's converted to the return type as defined
    in the python method.

    state is the value returned by handle_request
    """

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

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

class Metadata(WebArg):
    """
    Annotation for models that hold other WebArg fields.
    Can be used to group request parameters as an alternative to passing parameters directly.

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

Bases: Param

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

class Path(Param):
    def __init__(
        self,
        alias: typing.Optional[str] = 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

__init__(alias=None, /, *, style=SimpleString)

Parameters:

Name	Type	Description	Default
alias	Optional[str]	Path parameter name, if different than the name of the annotated parameter	None
style	type[StringSerializationStyle]	Serialization style	SimpleString

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

def __init__(
    self,
    alias: typing.Optional[str] = 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

Bases: Param

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

class Query(Param):
    def __init__(
        self,
        alias: typing.Optional[str] = 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

__init__(alias=None, /, *, style=FormExplode)

Parameters:

Name	Type	Description	Default
alias	Optional[str]	Query parameter name, if different than the name of the annotated parameter	None
style	type[MultimapSerializationStyle]	Serialization style	FormExplode

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

def __init__(
    self,
    alias: typing.Optional[str] = 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

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.

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

@dc.dataclass
class Responses(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.
    """

    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][lapidary.runtime.Body]
    """

responses: Mapping[StatusCodeRange, Response] instance-attribute

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

UnexpectedResponse

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

class UnexpectedResponse(LapidaryResponseError):
    """Raised when the remote server responded with code and content-type pair that wasn't declared in the method return annotation"""

    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)

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

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