api-examples.md

🎪 Clientele by example

All these examples can be copied and pasted into a file and run in a python application.

Simple GET request

from clientele import api

client = api.APIClient(base_url="https://pokeapi.co/api/v2")

@client.get("/pokemon/{pokemon_name}")
def get_pokemon_info(result: dict, pokemon_name: str) -> dict:
    return result

The simplest logic you can do with Clientele.
No validation will be ran on the data.

Receive specific data in result

from clientele import api
from pydantic import BaseModel

client = api.APIClient(base_url="https://pokeapi.co/api/v2")


class PokemonInfo(BaseModel):
    name: str
    id: int


@client.get("/pokemon/{pokemon_name}")
def get_pokemon_info(result: PokemonInfo, pokemon_name: str) -> PokemonInfo:
    return result

Use Pydantic BaseModel to return only the data you want in the result parameter.
Pydantic's model_validate will be ran against the response.json.
Only values explicitly declared in the BaseModel will be returned.

Return specific data after result

from clientele import api

client = api.APIClient(base_url="https://pokeapi.co/api/v2")

@client.get("/pokemon/{pokemon_name}")
def get_pokemon_info(result: dict, pokemon_name: str) -> str:
    return result.get("name")

The return type of the decorated function does not need to match the result parameter.
You can return whatever you like.
This is also a good time to do logging, persistence of results, dispatching post-request actions, etc.

Query parameters

Using parameters

from clientele import api

client = api.APIClient(base_url="https://pokeapi.co/api/v2")


@client.get("/pokemon/")
def get_pokemon_page(result: dict, limit: int, offset: int) -> dict:
    return result

get_pokemon_page(limit=10, offset=30)

Parameters not declared in the path string will instead become query parameters.
Optional or None values will be ignored.

Using query dict

from clientele import api

client = api.APIClient(base_url="https://pokeapi.co/api/v2")


@client.get("/pokemon/")
def get_pokemon_page(result: dict) -> dict:
    return result

get_pokemon_page(query={"limit": 10, "offset": 30})

You can pass a dict query to achieve the same results.
This does not need to be declared in your decorated function.

Using aliases

Some APIs use query parameter names that aren't valid, idiomatic Python identifiers (e.g. camelCase), which would trip up linters like ruff's N803. Use Annotated with api.Query(alias=...) to give the parameter a Python-friendly name while still sending the API's expected name over the wire:

from typing import Annotated

from clientele import api

client = api.APIClient(base_url="https://pokeapi.co/api/v2")


@client.get("/pokemon/")
def get_pokemon_page(result: dict, order_by: Annotated[str, api.Query(alias="orderBy")]) -> dict:
    return result

get_pokemon_page(order_by="name")

The function is called with order_by, but the request is sent as ?orderBy=name.
Clients generated from an OpenAPI spec add this automatically whenever a query parameter's name isn't already valid snake_case.

Simple POST request

from clientele import api

client = api.APIClient(base_url="https://httpbin.org")


@client.post("/post")
def post_input_data(result: dict, data: dict) -> dict:
    return result

The data parameter is serialized to JSON and sent in an HTTP POST request.
This pattern is identical in PUT / PATCH / DELETE decorators functions.

Data validation

from clientele import api
from pydantic import BaseModel

client = api.APIClient(base_url="https://httpbin.org")


class InputData(BaseModel):
    name: str
    email: str


@client.post("/post")
def post_input_data(result: dict, data: InputData) -> InputData:
    return result

Pydantic will run model_validate on the data parameter before sending the HTTP POST request.
This pattern is identical in PUT / PATCH / DELETE decorators functions.

Inspect HTTP responses

from clientele import api
import httpx

client = api.APIClient(base_url="https://pokeapi.co/api/v2")


@client.get("/pokemon/{pokemon_name}")
def get_pokemon_info(result: dict, response: httpx.Response, pokemon_name: str) -> dict:
    print(response.headers)
    return result

Pass the response parameter to the decorated function to receive the httpx.Response object.

Control response parsing

Using a callback

from clientele import api
import httpx

client = api.APIClient(base_url="https://pokeapi.co/api/v2")


def parse_response_myself(response: httpx.Response) -> dict:
    data = response.json()
    return {
        "my_custom_key": data["name"],
    }


@client.get("/pokemon/{pokemon_name}", response_parser=parse_response_myself)
def get_pokemon_info(result: dict, pokemon_name: str) -> str:
    return result["my_custom_key"]

Pass a callable to the response_parser parameter to control how http responses are parsed.
Clientele will no longer handle any data validation for you, but you have complete control.
The return type of this callback must match the type of the result parameter.

Using strong types

from clientele import api
import httpx
from pydantic import BaseModel

client = api.APIClient(base_url="https://pokeapi.co/api/v2")


class MyResult(BaseModel):
    custom_key: str


def parse_response_myself(response: httpx.Response) -> MyResult:
    data = response.json()
    return MyResult(
        custom_key=data["name"],
    )


@client.get("/pokemon/{pokemon_name}", response_parser=parse_response_myself)
def get_pokemon_info(result: MyResult, pokemon_name: str) -> str:
    return result.custom_key

Using a map

import httpx
from pydantic import BaseModel

from clientele import api

client = api.APIClient(base_url="https://pokeapi.co/api/v2")


class OkResult(BaseModel):
    name: str

class NotFoundResult(BaseModel):
    name: str


@client.get("/pokemon/{pokemon_name}", response_map={200: OkResult, 404: NotFoundResult})
def get_pokemon_info(result: OkResult | NotFoundResult, pokemon_name: str) -> str:
    return result.name

The response_map accepts {int: ResponseModel}.
The HTTP response status code will be matched against the model and used as validation.
If the http response does not match any status codes in the response_map then an clientele.api.APIException exception will be raised.

Handling errors

from pydantic import BaseModel

from clientele import api

client = api.APIClient(base_url="https://pokeapi.co/api/v2")


class OnlyErrorResult(BaseModel):
    name: str


@client.get("/pokemon/{pokemon_name}", response_map={500: OnlyErrorResult})
def get_pokemon_info(result: OnlyErrorResult, pokemon_name: str) -> str:
    return result.name


try:
    get_pokemon_info("pikachu")
except api.APIException as e:
    print(f"Error occurred: {e.reason}")
    print(f"Response details: {e.response}")

Unexpected response statuses will throw an clientele.api.APIException exception.
If there is no response_map provided then Clientele will call raise_for_status on the httpx.Response object.
If response_map is provided then the httpx.Response status code must match one of the keys.
The APIException will have a human readable reason.
The APIException will also have the httpx.Response that raised the exception for inspection.

Configuration

Using BaseConfig

from clientele import api
import httpx

my_config = api.BaseConfig(base_url="https://httpbin.org")

client = api.APIClient(config=my_config)


@client.get("/get")
def my_function(result: dict) -> dict:
    return result

Instead of providing base_url to APIClient you can instead provide a BaseConfig object.
This gives you simplified access to common http configuration options.

Custom headers

from clientele import api
import httpx

my_config = api.BaseConfig(
    base_url="https://httpbin.org", 
    headers={"Custom-Header": "Hello, Clientele!"}
)

client = api.APIClient(config=my_config)


@client.get("/get")
def return_headers(result: dict) -> str:
    """httpbin returns the headers it received."""
    return result["headers"]["Custom-Header"]

Headers can be configured through BaseConfig.
See full configuration options here.

Async

Make multiple requests in parallel

import asyncio
from clientele import api

client = api.APIClient(base_url="https://pokeapi.co/api/v2")


@client.get("/pokemon/{pokemon_id}")
async def get_pokemon_name(result: dict, pokemon_id: int) -> str:
    return result["name"]


async def gather():
    async_tasks = [get_pokemon_name(pokemon_id=i) for i in range(1, 152)]
    return await asyncio.gather(*async_tasks)


def get_all_pokemon_names():
    return asyncio.run(gather())

Use the common gather / run pattern to build modular API calls with Clientele.
This example executes 151 HTTP requests in parallel.

Caching

Simple caching example

from clientele import api, cache

client = api.APIClient(base_url="https://pokeapi.co/api/v2")

@cache.memoize(ttl=300)  # Cache for 5 minutes
@client.get("/pokemon/{pokemon_id}")
def get_pokemon(result: dict, pokemon_id: int) -> dict:
    return result

# First call - hits the API
pikachu = get_pokemon(pokemon_id=25)

# Second call - returns cached result (no HTTP request)
pikachu_cached = get_pokemon(pokemon_id=25)

Caching Paginated Results

from clientele import api, cache

client = api.APIClient(base_url="https://pokeapi.co/api/v2")

@cache.memoize(ttl=300)
@client.get("/pokemon")
def list_pokemon(result: dict, limit: int, offset: int) -> dict:
    return result

# Each limit/offset combination is cached separately
page1 = list_pokemon(limit=20, offset=0)
page2 = list_pokemon(limit=20, offset=20)
page1_again = list_pokemon(limit=20, offset=0)  # Cached!

Caching with Query Parameters

from clientele import api, cache

client = api.APIClient(base_url="https://pokeapi.co/api/v2")

@cache.memoize(ttl=600)
@client.get("/ability")
def list_abilities(result: dict, limit: int, offset: int) -> dict:
    return result

# Different query params = different cache entries
abilities1 = list_abilities(limit=10, offset=0)
abilities2 = list_abilities(limit=20, offset=0)  # Different cache entry
abilities3 = list_abilities(limit=10, offset=0)  # Uses cached abilities1

Namespace Isolation with Custom Keys

from clientele import api, cache

client = api.APIClient(base_url="https://pokeapi.co/api/v2")

@cache.memoize(
    ttl=300,
    key=lambda pokemon_id, version_id: f"pokemon:{pokemon_id}:version:{version_id}"
)
@client.get("/pokemon/{pokemon_id}")
def get_pokemon_version(result: dict, pokemon_id: int, version_id: int) -> dict:
    # Custom key ensures different versions are cached separately
    return result

# Each pokemon/version combination has its own cache entry
red_pikachu = get_pokemon_version(pokemon_id=25, version_id=1)
blue_pikachu = get_pokemon_version(pokemon_id=25, version_id=2)

Short-Lived Cache for Rate Limiting

from clientele import api, cache

client = api.APIClient(base_url="https://pokeapi.co/api/v2")

# Cache for just 10 seconds to reduce burst traffic
@cache.memoize(ttl=10)
@client.get("/pokemon/{pokemon_id}")
def get_pokemon_burst(result: dict, pokemon_id: int) -> dict:
    return result

# Multiple rapid calls within 10 seconds use cache
for _ in range(100):
    get_pokemon_burst(pokemon_id=25)  # Only makes 1 HTTP request

Streaming

Basic Server Sent Events example

from typing import AsyncIterator

from clientele import api
from pydantic import BaseModel


client = api.APIClient(base_url="https://httpbin.org")


class Event(BaseModel):
    id: int
    url: str


@client.get("/stream/{n}", streaming_response=True)
async def stream_events(result: AsyncIterator[Event], n: int) -> AsyncIterator[Event]:
    return result

async for event in await stream_events(n=4):
    print(event)

See Streaming for more.

Direct requests

from clientele import api
from pydantic import BaseModel

client = api.APIClient(base_url="https://pokeapi.co/api/v2")

class Pokemon(BaseModel):
    name: str
    height: int
    weight: int

# Direct GET request
result = client.request(
    "GET",
    "/pokemon/{pokemon_id}",
    response_map={200: Pokemon},
    pokemon_id=1
)

See direct requests for more.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Expand file tree

🎪 Clientele by example

Simple GET request

Receive specific data in result

Return specific data after result

Query parameters

Using parameters

Using query dict

Using aliases

Simple POST request

Data validation

Inspect HTTP responses

Control response parsing

Using a callback

Using strong types

Using a map

Handling errors

Configuration

Using BaseConfig

Custom headers

Async

Make multiple requests in parallel

Caching

Simple caching example

Caching Paginated Results

Caching with Query Parameters

Namespace Isolation with Custom Keys

Short-Lived Cache for Rate Limiting

Streaming

Basic Server Sent Events example

Direct requests

Search code, repositories, users, issues, pull requests...

FilesExpand file tree

api-examples.md

Latest commit

History

api-examples.md

File metadata and controls

🎪 Clientele by example

Simple GET request

Receive specific data in result

Return specific data after result

Query parameters

Using parameters

Using query dict

Using aliases

Simple POST request

Data validation

Inspect HTTP responses

Control response parsing

Using a callback

Using strong types

Using a map

Handling errors

Configuration

Using BaseConfig

Custom headers

Async

Make multiple requests in parallel

Caching

Simple caching example

Caching Paginated Results

Caching with Query Parameters

Namespace Isolation with Custom Keys

Short-Lived Cache for Rate Limiting

Streaming

Basic Server Sent Events example

Direct requests

Expand file tree