Server SDKs

Python examples

Async cancellation, serialization, and provider configuration.

Configure backups

import os
from affordable_geo_apis import (
    FallbackBudget,
    FallbackCircuitBreakerOptions,
    FallbackOptions,
    GeocodingClient,
)

with GeocodingClient(
    base_url=os.environ['GEOCODING_BASE_URL'],
    api_key=os.environ['GEOCODING_API_KEY'],
    confidence_threshold=0.8,
    fallback_budget=FallbackBudget(max_requests=1000),
    fallback_circuit_breaker=FallbackCircuitBreakerOptions(
        failure_threshold=3, cooldown_ms=30_000,
    ),
    fallbacks=[
        FallbackOptions(provider='google', api_key=os.environ['GOOGLE_MAPS_API_KEY']),
        FallbackOptions(provider='here', api_key=os.environ['HERE_API_KEY']),
    ],
) as client:
    response = client.geocode({
        'street': '10 Downing Street', 'city': 'London', 'country': 'GB',
    }, timeout=15)
    print(response.provider, response.fallback_reason)
    print(response.to_dict())

GB skips primary. The first nonempty configured backup wins. Selected backup metadata is available in response.provider_response. The async client accepts the same constructor options. Fallbacks, budget, and circuit options also accept dictionaries with the same snake_case field names.

Handle errors and deadlines

import os
from affordable_geo_apis import GeocodingClient, GeocodingError

with GeocodingClient(
    base_url=os.environ['GEOCODING_BASE_URL'],
    api_key=os.environ['GEOCODING_API_KEY'],
) as client:
    try:
        response = client.geocode({
            'street': '11 Wall Street', 'city': 'New York',
        }, timeout=15)
        if not response.results:
            print('No match')
    except TimeoutError:
        print('The total provider-chain deadline expired')
    except GeocodingError as error:
        print(error.provider, error.code, error.status)
        print(error.to_dict())

Per-call timeout is in seconds. Constructor timeout_ms is in milliseconds per provider. A provider timeout is GeocodingError(code='timeout') and can trigger fallback. A total deadline raises TimeoutError and stops the chain.

Async cancellation

import asyncio
import os
from affordable_geo_apis import AsyncGeocodingClient

async def main():
    async with AsyncGeocodingClient(
        base_url=os.environ['GEOCODING_BASE_URL'],
        api_key=os.environ['GEOCODING_API_KEY'],
    ) as client:
        task = asyncio.create_task(client.geocode({
            'street': '11 Wall Street', 'city': 'New York',
        }))
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            print('Request cancelled')

asyncio.run(main())

An enclosing asyncio.timeout(...) can also bound calls. Cancellation stops the chain and is not converted to a provider failure.

Serialization and reuse

import json

# After obtaining a response:
# print(json.dumps(response.to_dict()))

to_dict() serializes nested results, attempts, and errors. It omits provider_response for primary results; otherwise the entire selected backup body is retained. Python model and serialized response fields use snake_case.

The blocking client uses a private event loop in a background thread and can serve concurrent threads. The async client belongs to one event loop. Reuse clients for pooling, budgets, and circuits; close them at application shutdown.

Custom networking

Pass transport=httpx.MockTransport(handler) to either client for mocked requests, or supply another HTTPX async transport. The client owns and closes the transport. Handlers may be synchronous or async; custom transports must cooperate with asyncio cancellation and must not block the event loop.

Address autocomplete

Blocking client

import os
from affordable_geo_apis import GeocodingClient

with GeocodingClient(
    base_url=os.environ['GEOCODING_BASE_URL'],
    api_key=os.environ['GEOCODING_API_KEY'],
) as client:
    response = client.autocomplete({'q': '11 Wall', 'country': 'US'}, timeout=15)
    if response.suggestions:
        address = client.resolve_autocomplete(response.suggestions[0], timeout=15)
        print(address.label, address.lat, address.lon)

Async client

import asyncio
import os
from affordable_geo_apis import AsyncGeocodingClient

async def main():
    async with AsyncGeocodingClient(
        base_url=os.environ['GEOCODING_BASE_URL'],
        api_key=os.environ['GEOCODING_API_KEY'],
    ) as client:
        response = await client.autocomplete({'q': '11 Wall', 'country': 'US'}, timeout=15)
        if response.suggestions:
            address = await client.resolve_autocomplete(response.suggestions[0], timeout=15)
            print(address.label, address.lat, address.lon)

asyncio.run(main())

Keep the suggestion object, including its provider and Google session token, until selection. For a Google-backed typing session, reuse session_token across searches and generate a new token for the next session. It is SDK-only and omitted from primary HTTP requests. Resolution is a separate provider call outside fallback budgets/circuit breakers; it returns one address without an envelope or attempt history. Both per-call timeout values above are in seconds. See the autocomplete guide.