Skip to content

Client

AlmaClient is the entry point. It is a plain async class – no framework dependency – composing a niquests.AsyncSession for transport, a TokenBucket for rate limiting, an AdaptiveController for backpressure, and an asyncio.Semaphore to cap concurrency.

Use it as an async context manager so the underlying session is closed:

async with AlmaClient("your-api-key") as client:
    ...

Most of what you will use it for is the namespaces it exposes – client.users, client.bibs, client.acq, client.config, client.analytics, client.primo – each documented on its own page.

AlmaClient 

AlmaClient(
    apikey: str,
    location: Literal[
        "America",
        "Europe",
        "Asia Pacific",
        "Canada",
        "China",
    ] = "Europe",
    *,
    rate_limit: float = 25.0,
    concurrent_requests: int = 150,
    retry_attempts: int = 3,
    backoff_factor: float = 0.5,
    recovery_increment: float = 1.0,
    recovery_window: float = 10.0,
    cooldown: float = 5.0,
    max_wait: float | None = None,
    client: AsyncSession | None = None,
)

Async API wrapper for the Alma library management system.

Composed of: niquests.AsyncSession (transport), TokenBucket (rate limiting), AdaptiveController (AIMD backpressure), asyncio.Semaphore (concurrency cap).

execute async 

execute(
    method: str,
    url: str,
    *,
    parser: Parser,
    model: type[_ModelT],
    validate: Callable[[Response], None] = ...,
    retry: bool | None = ...,
    **kwargs: Any,
) -> _ModelT

execute(
    method: str,
    url: str,
    *,
    parser: Parser,
    model: None = ...,
    validate: Callable[[Response], None] = ...,
    retry: bool | None = ...,
    **kwargs: Any,
) -> RESP_TYPE

Send one request through the rate limiter, retry loop and error mapping.

You are not normally expected to call this. Every namespace method – client.users.get_user(), client.bibs.get_item() and the rest – is a thin wrapper around it that supplies the URL, the parser and the response type. Prefer those: they are typed, and they name the thing you are asking for. This is the shared chokepoint they all funnel through, and it is public so that the two cases below remain possible.

Call it directly when:

  • Alma exposes an endpoint almapy does not wrap yet. Going through execute keeps the throttling, retries, backpressure and error mapping that raw HTTP would lose.
  • You need to override the retry policy for a single request, with retry=. The namespace methods do not accept that argument.

Parameters:

Name Type Description Default
method str

HTTP verb, e.g. "GET" or "POST".

 required
url str

Path relative to the regional gateway's /almaws/v1 root, e.g. "/users/12345678". AlmaEndpoint.build() produces these and percent-encodes the path parameters.

 required
parser Parser

How to decode the response body – "json" for a Box, "xml" for parsed XML, "text" for a raw str (MARC XML), "none" for an empty body.

 required
model Any

Optional Pydantic model class to validate the response into, exactly as on the namespace methods.

 None
validate Callable[[Response], None]

Response validator, run on every response. Defaults to the one mapping Alma's error codes onto almapy.exceptions; override only to opt out of that mapping.

 _validate_response
retry bool | None

True forces full retries even for a write – only where a duplicate would be harmless. False disables retries for this request. None (the default) decides by method idempotency, so POST and PATCH are not replayed on ambiguous failures. See Rate limiting.

 None
**kwargs Any

Passed to the underlying niquests call – params, json, data, headers. A json body is serialised once, before the retry loop, so dump/model_dump objects work here as they do on the namespace methods.

 {}

Returns:

Type Description
Any

The parsed body: a Box for parser="json", a str for
Any

parser="text", or an instance of model when one is given.

Raises:

Type Description
APIClientError

For 4xx responses, or a more specific subclass where Alma's error code maps to one.
APIServerError

For 5xx responses that survived the retries.
MalformedResponseError

For a 2xx whose body could not be parsed as the requested format – an HTML maintenance page, say.
ThrottleTimeoutError

If max_wait is configured and elapsed while waiting for a rate-limit token.

Examples:

Reaching an endpoint almapy does not wrap:

resp = await client.execute(
    "GET",
    "/task-lists/requested-resources",
    parser="json",
    params={"library": "MAIN", "circ_desk": "DEFAULT"},
)

aclose async 

aclose() -> None

Close the underlying HTTP client if this instance owns it.