Python SDK¶

Official Python client for the Climate-Lama API. Engine-agnostic — the SDK only talks HTTP to the backbone.

Install¶

pip install climate-lama-client

The package name on PyPI is climate-lama-client and the import path is climate_lama_client — they match by design. climate-lama is the backbone service, a different distribution (issue #528).

Quickstart¶

from climate_lama_client import ClimateLama

client = ClimateLama(base_url="https://api.example.com")
client.auth.login(email="you@example.com", password="...", org_slug="default")

hz = client.hazards.list()
ex = client.exposures.upload("portfolio.csv", name="My portfolio")

# exposure_id is optional — when omitted, the backbone uses the caller-org's
# most recent exposure dataset.
result = client.compute.impact(
    hazard_id=hz[0]["id"],
    exposure_id=ex["id"],
).wait(timeout_s=300, poll_interval_s=2)
print(result["ead"], result["aai"])

job.wait() polls GET /v1/jobs/{id} every poll_interval_s (default 2s). When the job completes, it auto-fetches GET /v1/results/{result_id} and returns the result summary. Raises JobError on failure and TimeoutError after timeout_s (default 300s).

Authentication¶

Compute endpoints require Role.ANALYST or above. Either log in via client.auth.login(...) (sets the bearer token on the client for subsequent calls), or pass a token at construction:

client = ClimateLama(base_url=..., token="<jwt-or-api-key>")

API keys (prefix clk_) and JWTs are both accepted as bearer tokens.

Cost-benefit¶

Cost-benefit runs against an existing impact result, not raw inputs. Capture the impact job's result_id first:

impact_job = client.compute.impact(hazard_id=..., exposure_id=...)
impact_job.wait()
result_id = impact_job.payload["result_id"]

cb = client.compute.cost_benefit(
    impact_result_id=result_id,
    measure_ids=[...],
    discount_rate=0.014,
    time_horizon_years=10,
    value_growth_rate=0.013,
).wait()

Scenario matrix (ORSA)¶

compute.impact_matrix() submits a scenario × horizon grid as a single batch and returns a Batch handle with the same wait()/refresh() shape as a job:

batch = client.compute.impact_matrix(
    hazard_id="hz-1",
    exposure_id="ex-1",
    scenario_labels=["baseline", "ssp2-4.5", "ssp5-8.5"],
    horizon_years=[2030, 2050, 2080],
)
batch.wait(timeout_s=300, poll_interval_s=3)

for cell in batch.cells:
    if cell["result_id"]:
        print(cell["scenario_label"], cell["horizon_year"], client.results.get(cell["result_id"]))
    else:
        print(cell["scenario_label"], cell["horizon_year"], cell["status"])

Cells run independently, and a batch has no failure state — it reaches completed once every cell's job is terminal, whether or not each cell succeeded. wait() therefore never raises JobError; check each cell's status/error_message. TimeoutError is still raised after timeout_s. client.compute.get_impact_matrix(batch_id) is the raw getter for reattaching to a batch from a prior run.

See sdk/python/examples/orsa_matrix.py for a runnable walkthrough.

Field-name translation¶

The SDK uses friendlier hazard_id / exposure_id parameter names but translates them to the backbone's wire-level hazard_dataset_id / exposure_dataset_id before sending. Cost-benefit's wire shape is mostly 1:1 with the SDK because there's no shorter friendly name for impact_result_id.

Errors¶

The backbone returns ADR-028 error envelopes (code, severity, message, details). The SDK decodes them into typed Python exceptions, one per registered code:

from climate_lama_client.errors import (
    HazardIntensityUnitMismatchError,
    NotFoundError,
    ClimateLamaError,
)

try:
    client.compute.impact(hazard_id="hz-1", exposure_id="ex-1").wait()
except HazardIntensityUnitMismatchError as exc:
    # Catch a specific code…
    print(exc.message, exc.details)
except ClimateLamaError as exc:
    # …or fall back to the base class for anything you didn't enumerate.
    print("API error", exc.code, exc.message)

The error class registry is generated from the backbone's src/climate_lama/core/errors.py by sdk/python/scripts/regenerate_errors.py, so the SDK never drifts from the backbone's published codes. Codes the SDK doesn't recognize fall back to the base ClimateLamaError class instead of crashing — which is what makes it safe to add new codes to the backbone without breaking older SDK versions.

Async support¶

AsyncClimateLama is the asyncio twin of ClimateLama, built on httpx.AsyncClient. It lives in climate_lama_client.aio and is also re-exported from the package root:

import asyncio
from climate_lama_client.aio import AsyncClimateLama


async def main() -> None:
    async with AsyncClimateLama(base_url="https://api.example.com") as client:
        await client.auth.login(email="you@example.com", password="...", org_slug="default")

        hz = await client.hazards.list()
        ex = await client.exposures.upload("portfolio.csv", name="My portfolio")

        job = await client.compute.impact(hazard_id=hz[0]["id"], exposure_id=ex["id"])
        result = await job.wait(timeout_s=300, poll_interval_s=2)
        print(result["ead"], result["aai"])


asyncio.run(main())

The two clients are at feature parity — same resources (auth, hazards, exposures, compute, jobs, results), same method names, same arguments, same typed errors. There are exactly two differences:

every request-issuing method is a coroutine and must be awaited (set_token() stays synchronous — it only mutates headers);
close() becomes aclose(), following the httpx.AsyncClient convention. async with handles it for you.

AsyncJob.wait() and AsyncBatch.wait() poll with asyncio.sleep, so a waiting job yields the event loop instead of blocking it. That is what makes it worth running several at once:

jobs = await asyncio.gather(*[client.compute.impact(hazard_id=h["id"]) for h in hz])
results = await asyncio.gather(*[job.wait() for job in jobs])

Which client should I use?¶

Use ClimateLama in scripts, notebooks, and anything already synchronous — it is not deprecated and never will be. Use AsyncClimateLama inside an existing event loop (FastAPI handlers, async workers) or when you want many concurrent jobs in flight from one process.

They are separate classes rather than one client with both modes, so a type checker can tell you that you forgot an await. Everything transport- agnostic — envelope decoding, the friendly-to-wire field-name translation, terminal-status classification — is literally the same code behind both, so the surfaces cannot drift.

Versioning¶

The SDK is shipped from the sdk/python/ subdirectory of the climate-lama repo. Tagging a release as sdk-vX.Y.Z publishes to Test PyPI automatically; production PyPI publishes are gated behind a manual approval step.

Source¶

sdk/python/ on GitHub.