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	Base class for HTTP middleware.
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})

Cookie

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

Bases: Param

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

Source code in 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

Header

Header(alias: Optional[str] = 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: Optional[str] DEFAULT: None
style	Serialization style TYPE: type[MultimapSerializationStyle] DEFAULT: SimpleMultimap

Source code in 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

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: Optional[str] = None, /, *, style: type[StringSerializationStyle] = SimpleString)

Bases: Param

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

Source code in 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

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

Bases: Param

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

Source code in 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

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(security: Iterable[SecurityRequirements] | None = None, session_factory: SessionFactory = AsyncClient, middlewares: Sequence[HttpxMiddleware] = (), **httpx_kwargs: Unpack[ClientArgs])

Bases: ABC

Base for Client classes

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

METHOD	DESCRIPTION
lapidary_authenticate	Register named Auth instances for future use with methods that require authentication.
lapidary_deauthenticate	Remove reference to a given Auth instance.

Source code in 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

lapidary_authenticate(*auth_args: NamedAuth, **auth_kwargs: 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

Source code in 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

lapidary_deauthenticate(*sec_names: str) -> None

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

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

HttpxMiddleware

Bases: Generic[State]

Base class for HTTP middleware.

METHOD	DESCRIPTION
handle_request	Called for each request after it's generated for a method call but before it's sent to the remote server.
handle_response	Called for each response after it's been received from the remote server and before it's converted to the return type as defined

handle_request abstractmethod async

handle_request(request: 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.

Source code in 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 async

handle_response(response: Response, request: 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

Source code in 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
    """

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

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

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

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], Optional[C]]

Source code in 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