Error handling

The Python SDK returns failures as typed values instead of raising them, so every operation call yields a union you branch on rather than a try/except you wrap around it.

Every Geopera operation is a POST that resolves to one of a small, fixed set of outcomes — a success model, a Problem, an HTTPValidationError, or None — and the unhappy path is part of the return type, not a separate exception hierarchy. There is exactly one SDK-defined exception class, and it only fires for status codes the operation’s contract does not describe. This page covers the union you get back from the fluent Geopera client, the same union from the low-level sync/asyncio functions, the status-aware *_detailed and detailed=True variants, the Problem and HTTPValidationError models field by field, and the one exception (errors.UnexpectedStatus) plus the timeout that propagates from httpx. For the wire-level error contract shared by every client, see Errors.

Errors are returned, not raised

A call on the fluent client returns either the operation’s success model or one of the documented error models — nothing is thrown for a documented failure:

from geopera import Geopera

client = Geopera(token="gpra_...")

result = client.catalog.search({
    "host_name": "earthsearch-aws",
    "collections": ["sentinel-2-l2a"],
    "limit": 10,
})
# result is: CatalogSearchOutput | Problem | HTTPValidationError | None

The SDK inspects the HTTP status and parses the body into the matching model. The mapping is identical for every operation:

Because nothing is raised for a documented error, you branch on the runtime type with isinstance:

from geopera import Geopera
from geopera.models import CatalogSearchOutput, Problem, HTTPValidationError

client = Geopera(token="gpra_...")

result = client.catalog.search({
    "host_name": "earthsearch-aws",
    "collections": ["sentinel-2-l2a"],
    "limit": 10,
})

if isinstance(result, CatalogSearchOutput):
    print("ok:", result)
elif isinstance(result, Problem):
    print("geopera error:", result.status, result.title, "-", result.detail)
elif isinstance(result, HTTPValidationError):
    print("invalid input:", result.detail)
elif result is None:
    print("undocumented response (no body parsed)")

The branches above are exhaustive for the union. Checking the success model first is the idiomatic pattern; the remaining arms tell Problem (a server-side or permission error) apart from HTTPValidationError (your request body failed validation), and None covers an undocumented status when raising is off.

The same union from the low-level functions

The fluent client is a thin wrapper over the generated operation modules, and they return the exact same union. Drop to the low-level form when you need the async variants or want to construct the typed input model explicitly:

from geopera import AuthenticatedClient
from geopera.api.operations import catalog_search
from geopera.models import CatalogSearchInput, CatalogSearchOutput, Problem, HTTPValidationError

client = AuthenticatedClient(base_url="https://api.geopera.com", token="gpra_...")

result = catalog_search.sync(
    client=client,
    body=CatalogSearchInput(
        host_name="earthsearch-aws",
        collections=["sentinel-2-l2a"],
        limit=10,
    ),
)
# Same type as the fluent call:
# CatalogSearchOutput | HTTPValidationError | Problem | None

sync() is defined as sync_detailed(...).parsed — the parsing logic lives in one place, so the fluent client, sync, and asyncio all surface failures the same way. You can reach the underlying client from the fluent object via client.client (an AuthenticatedClient) if you want to mix the two styles.

The Problem model

Problem is the RFC 7807 / 9457 problem+json object returned for 401, 403, 404, and 500. It carries four declared fields plus a bag for any extension members:

Note that the Python attribute is type_ (trailing underscore) because type is a builtin; the JSON key on the wire is type. title, status, and detail are always present on a parsed Problem. type_ defaults to "about:blank" and is technically str | Unset, so guard it if you compare against problem-type URIs.

Any extra members the server includes — such as instance or a domain-specific code — land in additional_properties and are reachable by subscript (Problem implements __getitem__, __contains__, and exposes additional_keys):

if isinstance(result, Problem):
    print(result.status)          # e.g. 403
    print(result.title)           # e.g. "Forbidden"
    print(result.detail)          # human-readable explanation
    print(result.type_)           # problem-type URI, or "about:blank"

    for key in result.additional_keys:   # every extension member
        print(key, "=", result[key])

    if "code" in result:          # membership test on the extension bag
        print("error code:", result["code"])

result.status mirrors the HTTP status, so you can switch on it without reaching for the detailed response:

if isinstance(result, Problem):
    if result.status == 401:
        raise RuntimeError("token rejected — check your gpra_ key or session token")
    elif result.status == 403:
        raise RuntimeError(f"missing scope: {result.detail}")
    elif result.status == 404:
        print("not found")
    elif result.status == 500:
        print("server error, retry later")

Which scopes a given operation requires is documented under Scopes; 401 vs 403 semantics and the canonical problem-type URIs are described in Errors.

The HTTPValidationError model

A 422 means the request body did not satisfy the operation’s input schema. The SDK returns HTTPValidationError, whose single detail field is a list of ValidationError items. detail is typed list[ValidationError] | Unset — it is Unset if the server returns no list — so always guard with isinstance(..., Unset) before iterating.

Each ValidationError has three declared fields:

As with Problem, the JSON key is type and the Python attribute is type_. Iterate detail to report each field error:

from geopera.types import Unset

if isinstance(result, HTTPValidationError):
    if isinstance(result.detail, Unset):
        print("validation failed (no field detail returned)")
    else:
        for err in result.detail:
            field = ".".join(str(part) for part in err.loc)
            print(f"{field}: {err.msg} ({err.type_})")

A typical 422 body looks like this on the wire:

{
	"detail": [
		{
			"loc": ["body", "limit"],
			"msg": "Input should be less than or equal to 100",
			"type": "less_than_equal"
		}
	]
}

Validation failures are deterministic — the same body always fails the same way — so they should be fixed in code, not retried. A common cause is passing a plain dict with a misspelled or out-of-range field; constructing the typed input model (CatalogSearchInput(...)) instead lets your type checker catch many of these before the request is ever sent.

Reading the status code, headers, and bytes

When you need the raw HTTP status, response headers, or the response body, ask for the detailed response. On the fluent client pass detailed=True; on the low-level functions call sync_detailed (or asyncio_detailed). Either way you get a Response[...] object instead of the bare parsed value:

from http import HTTPStatus
from geopera import Geopera
from geopera.models import CatalogSearchOutput, Problem, HTTPValidationError

client = Geopera(token="gpra_...")

resp = client.catalog.search(
    {"host_name": "earthsearch-aws", "collections": ["sentinel-2-l2a"]},
    detailed=True,
)

if resp.status_code == HTTPStatus.OK:                       # 200
    data = resp.parsed                                       # CatalogSearchOutput
    print(data)
elif resp.status_code == HTTPStatus.UNPROCESSABLE_ENTITY:   # 422
    assert isinstance(resp.parsed, HTTPValidationError)
    print("invalid input:", resp.parsed.detail)
else:
    assert isinstance(resp.parsed, Problem)
    print(resp.status_code, resp.parsed.detail)

The low-level equivalent is catalog_search.sync_detailed(client=..., body=...), which returns the same Response[...]. status_code is an HTTPStatus enum, so it compares cleanly against both the enum members and plain integers (resp.status_code == 200). The headers mapping is where you read request-correlation and throttling metadata — see Rate limits. parsed is still None for an undocumented status (when raising is off), so check status_code first if you opened the detailed response specifically to inspect an unexpected code.

UnexpectedStatus and timeouts

There is exactly one exception class defined by the SDK, in geopera.errors:

class UnexpectedStatus(Exception):
    """Raised when the response has an undocumented status and raise_on_unexpected_status is True."""
    def __init__(self, status_code: int, content: bytes): ...

It is only raised when the server returns a status code that the operation’s contract does not describe (anything other than 200/401/403/404/422/500 for catalog.search, for example) and the client was constructed with raise_on_unexpected_status=True. The default is False, in which case an undocumented status makes the call return None and parsed on the detailed response None — no exception.

The flag is a constructor keyword on Client / AuthenticatedClient, and the fluent Geopera constructor forwards any extra keyword arguments straight through to the underlying client, so you set it the same way in either style:

import httpx
from geopera import Geopera, errors

# raise_on_unexpected_status is forwarded to the underlying AuthenticatedClient
client = Geopera(token="gpra_...", raise_on_unexpected_status=True)

try:
    result = client.catalog.search({
        "host_name": "earthsearch-aws",
        "collections": ["sentinel-2-l2a"],
    })
except errors.UnexpectedStatus as exc:
    print("undocumented status:", exc.status_code)
    print("raw body:", exc.content.decode(errors="ignore"))
except httpx.TimeoutException:
    print("request exceeded the client timeout")

errors.UnexpectedStatus exposes status_code: int and content: bytes; its str() message includes both the code and the decoded body. The only other thing that propagates as an exception is httpx.TimeoutException, raised by the underlying HTTP client when a request exceeds the configured timeout — this happens regardless of the raise_on_unexpected_status setting. Documented errors (401/403/404/422/500) never raise; they are always returned.

You configure the timeout at construction time, or derive a new client with a different budget via with_timeout:

import httpx
from geopera import AuthenticatedClient

# 30s connect+read budget for slow tasking/archive operations
client = AuthenticatedClient(
    base_url="https://api.geopera.com",
    token="gpra_...",
    timeout=httpx.Timeout(30.0),
)

# Or evolve an existing client (returns a new client; the original is unchanged)
patient = client.with_timeout(httpx.Timeout(60.0))

httpx.TimeoutException is the base for the finer-grained httpx.ConnectTimeout, httpx.ReadTimeout, and httpx.WriteTimeout; catching the base covers all of them. Network-level failures (DNS, refused connection, TLS) surface as httpx.ConnectError and other httpx.TransportError subclasses — neither the SDK nor the operation contract translates these into Problem values, so handle them as ordinary exceptions if you need to.

Async error handling

The async functions return the identical union, so the isinstance and detailed-response patterns carry over unchanged — only the call is awaited. The fluent client uses the synchronous transport, so reach for the low-level asyncio / asyncio_detailed functions when you need async/await:

import asyncio
import httpx
from geopera import AuthenticatedClient
from geopera import errors
from geopera.api.operations import catalog_search
from geopera.models import CatalogSearchInput, CatalogSearchOutput, Problem, HTTPValidationError

async def main() -> None:
    async with AuthenticatedClient(
        base_url="https://api.geopera.com",
        token="gpra_...",
    ) as client:
        try:
            result = await catalog_search.asyncio(
                client=client,
                body=CatalogSearchInput(
                    host_name="earthsearch-aws",
                    collections=["sentinel-2-l2a"],
                ),
            )
        except errors.UnexpectedStatus as exc:
            print("undocumented status:", exc.status_code)
            return
        except httpx.TimeoutException:
            print("timed out")
            return

        if isinstance(result, CatalogSearchOutput):
            print(result)
        elif isinstance(result, Problem):
            print("geopera error:", result.status, result.detail)
        elif isinstance(result, HTTPValidationError):
            print("invalid input:", result.detail)

asyncio.run(main())

asyncio_detailed mirrors sync_detailed, returning a Response[...] with status_code, parsed, headers, and content. Using the client as an async with context manager ensures the underlying httpx.AsyncClient is closed.

Gotcha: per-operation success models can carry their own errors list

The returned union is about the transport outcome. A 200 success model is still free to report partial, business-level failures inside its own body. For example, the archive- and tasking-estimate operations return a success model whose errors field is a list of structured items (ArchiveEstimateOutputErrorsItem, TaskingEstimateOutputErrorsItem) describing per-item problems — a quote that succeeded overall but could not price one host, say. These are not Problem objects and they do not change the HTTP status; you read them off the success model after the isinstance check passes:

from geopera import Geopera
from geopera.models import ArchiveEstimateOutput
from geopera.types import Unset

client = Geopera(token="gpra_...")

result = client.orders.archive.estimate({...})

if isinstance(result, ArchiveEstimateOutput):
    # `errors` is `list[...] | Unset`, so guard before iterating
    if not isinstance(result.errors, Unset):
        for item in result.errors:      # success, but with per-item warnings
            print("could not price:", item)

So a complete error strategy has two layers: branch on the returned union for transport-level outcomes, then inspect any success-model errors/warnings fields for partial failures.

Worked example: a reusable unwrap helper

Because every operation shares the same success | Problem | HTTPValidationError | None shape, you can centralise handling in one helper that turns returned errors into raised exceptions when that fits your app’s control flow better:

from typing import TypeVar
from geopera.models import Problem, HTTPValidationError
from geopera.types import Unset

T = TypeVar("T")


class GeoperaProblem(Exception):
    def __init__(self, problem: Problem):
        self.problem = problem
        super().__init__(f"{problem.status} {problem.title}: {problem.detail}")


class GeoperaValidationError(Exception):
    def __init__(self, error: HTTPValidationError):
        self.error = error
        fields = []
        if not isinstance(error.detail, Unset):
            fields = [
                f'{".".join(str(p) for p in e.loc)}: {e.msg}'
                for e in error.detail
            ]
        super().__init__("; ".join(fields) or "validation failed")


def unwrap(result: T | Problem | HTTPValidationError | None) -> T:
    """Return the success model, or raise a typed exception for any error."""
    if isinstance(result, Problem):
        raise GeoperaProblem(result)
    if isinstance(result, HTTPValidationError):
        raise GeoperaValidationError(result)
    if result is None:
        raise RuntimeError("undocumented response (no body parsed)")
    return result

Use it to collapse the union back into a single happy-path value:

from geopera import Geopera
from geopera.models import CatalogSearchOutput

client = Geopera(token="gpra_...")

try:
    output: CatalogSearchOutput = unwrap(
        client.catalog.search({
            "host_name": "earthsearch-aws",
            "collections": ["sentinel-2-l2a"],
        })
    )
    print(output)
except GeoperaProblem as exc:
    if exc.problem.status == 403:
        print("missing scope:", exc.problem.detail)
    else:
        raise
except GeoperaValidationError as exc:
    print("fix your request:", exc)

This keeps the typed-return contract at the boundary while giving the rest of your code a conventional exception flow. For idempotent retries on transient Problems (such as a 500), pair this with an idempotency key — see Idempotency.

See also

• Errors — the wire-level problem+json contract and problem-type URIs shared by all clients.
• Scopes — which scopes each operation requires (the source of most 403s).
• Authentication — gpra_ API keys and session tokens (the source of most 401s).
• Rate limits — throttling and the headers exposed on Response.