xAPI-client 1.0.11
A lightweight, flexible asynchronous API client for Python built on httpx and pydantic
Requires Python: >=3.12
Keywords: api, client, async, httpx, pydantic, rest, api-client, python-api-client

# xAPI

A lightweight, flexible asynchronous API client for Python built on [httpx](https://www.python-httpx.org/) and [pydantic](https://docs.pydantic.dev/).

xAPI organizes API endpoints into a tree of **Resources** and **Endpoints**, giving you a clean, dot-notation interface for calling any REST API with full type safety and automatic response validation.


```python
coin = await client.coins.coin(parameters=CoinParams.Bitcoin)
print(coin.name)  # "Bitcoin"
```

---

## Features

- **Async-first** &mdash; Built on `httpx.AsyncClient` for high-performance, non-blocking requests.
- **Resource-oriented** &mdash; Organize endpoints into a hierarchical tree. Access them with dot notation (`client.coins.coin(...)`).
- **Type-safe responses** &mdash; Pydantic models validate and parse every API response automatically.
- **List responses** &mdash; Handle endpoints that return JSON arrays by passing a `list[Model]` response type.
- **Parameterized paths** &mdash; Define URL templates like `{id}` and inject values with type-safe `Parameters` enums.
- **Query parameters** &mdash; Build and manage query strings with the `Query` class.
- **Authentication** &mdash; Scoped API key auth &mdash; apply globally, per-endpoint, or disable entirely.
- **Rate limiting** &mdash; Built-in sliding window rate limiter to stay within API quotas.
- **Retry with backoff** &mdash; Automatic exponential backoff on 5xx errors, timeouts, and connection failures. 4xx errors are raised immediately.
- **Structured logging** &mdash; Color-coded, per-component logging via `loguru` (enabled with `debug=True`).
- **Context manager** &mdash; Proper connection cleanup with `async with` support.

**NOTE:** This is an experimental project and proof of concept. May not fully work as indended.

---

## Installation

Install **xAPI** using pip

```shell
$ pip install xAPI
```
**Requirements:** Python 3.12+

---

## Quick Start

```python
import asyncio
import xAPI

from pydantic import BaseModel

# 1. Define a response model
class CoinModel(xAPI.ResponseModel):
    id: str
    symbol: str
    name: str

# 2. Define path parameters
class CoinParams(xAPI.Parameters):
    Bitcoin = "bitcoin"
    Ethereum = "ethereum"

# 3. Define a path with parameter placeholders
class CoinPath(xAPI.Path[CoinParams]):
    endpointPath: str = "{id}"

async def main():
    # 4. Create the client
    auth = xAPI.APIKey(
        key="x_cg_demo_api_key",
        secret="YOUR_API_KEY",
        scope="All",
        schem="Header"
    )

    async with xAPI.Client(
        url="https://api.example.com/v1/",
        apiKey=auth
    ) as client:

        # 5. Build resources and endpoints
        coins = xAPI.Resource("coins")
        coins.addEndpoints(
            xAPI.Endpoint(name="coin", path=CoinPath(), response=CoinModel)
        )
        client.add(coins)

        # 6. Make the request
        coin = await client.coins.coin(parameters=CoinParams.Bitcoin)
        print(coin.name)    # "Bitcoin"
        print(coin.symbol)  # "btc"

asyncio.run(main())
```

---

## Core Concepts

### Client

The `Client` is the entry point. It manages the HTTP connection, authentication, rate limiting, retries, and the resource tree.

```python
client = xAPI.Client(
    url="https://api.example.com/v1/",
    apiKey=auth,  # optional
    timeout=httpx.Timeout(30, connect=5),  # optional (default: 30s overall, 5s connect)
    rateLimit=xAPI.RateLimiter(maxCalls=30, perSecond=60),  # optional
    retry=xAPI.RetryConfig(attempts=3, baseDelay=0.3, maxDelay=5.0),  # optional
    debug=True, # optional, enables logging
)
```

Always close the client when done, either with `async with` or by calling `await client.close()`.

### Resource

A `Resource` is a named group of endpoints. Resources are registered on the client and accessed as attributes.

```python
coins = xAPI.Resource("coins")
client.add(coins)

# Now accessible as:
client.coins
```

**Path protection:** By default, the resource name is prepended to endpoint URLs. Set `pathProtection=False` to use the endpoint path as-is from the API root:

```python
# With pathProtection=True (default):
#   Endpoint path "list" -> request to /coins/list
coins = xAPI.Resource("coins")

# With pathProtection=False:
#   Endpoint path "global" -> request to /global
markets = xAPI.Resource("markets", pathProtection=False)
```

**Sub-resources** can be nested:

```python
parent = xAPI.Resource("api")
child = xAPI.Resource("coins")
parent.addResources(child)
# Access: client.api.coins.some_endpoint(...)
```

**Authentication scoping:** Set `requireAuth=True` on a resource to enable per-endpoint authentication (when using `Scope.Endpoint`).

### Endpoint

An `Endpoint` represents a single API call. It defines the HTTP method, URL path, response model, and validation behavior.

```python
endpoint = xAPI.Endpoint(
    name="coin",                  # Python attribute name
    path=CoinPath(),              # Path object with URL template
    response=CoinModel,           # Pydantic model or list[Model] for response parsing
    method="GET",                 # HTTP method (default: "GET")
    nameOverride="",              # Override the API-facing name
    strict=False,                 # Enable strict Pydantic validation
)
```

Add endpoints to a resource:

```python
resource.addEndpoints(endpoint)
# or multiple:
resource.addEndpoints([endpoint1, endpoint2, endpoint3])
```

Call an endpoint:

```python
# Simple endpoint (no path parameters)
result = await client.coins.list()

# With path parameters
result = await client.coins.coin(parameters=CoinParams.Bitcoin)

# With query parameters
query = xAPI.Query({"localization": False, "tickers": False})
result = await client.coins.coin(parameters=CoinParams.Bitcoin, query=query)
```

### Path & Parameters

Paths define URL templates. Parameters are typed enums that fill in the template placeholders.

```python
# Define parameters as a StrEnum
class CoinParams(xAPI.Parameters):
    Bitcoin = "bitcoin"
    Ethereum = "ethereum"
    Solana = "solana"

# Define a path with a placeholder
class CoinByID(xAPI.Path[CoinParams]):
    endpointPath: str = "{id}"

# Path without parameters
class CoinList(xAPI.Path):
    endpointPath: str = "list"

# Multi-segment path
class CoinTickers(xAPI.Path[CoinParams]):
    endpointPath: str = "{id}/tickers"
```

### Query

The `Query` class manages URL query parameters. Values set to `"NOT_GIVEN"` are automatically filtered out.

```python
query = xAPI.Query({
    "vs_currency": "usd",
    "order": "market_cap_desc",
    "per_page": 100,
    "sparkline": False,
    "optional_param": "NOT_GIVEN",  # filtered out
})

# Add more params
query.add({"page": 2})

# Remove a param
query.remove("sparkline")

# Inspect
print(query.queries)       # dict of active params
print(query.queryString)   # "vs_currency=usd&order=market_cap_desc&..."
```

### ResponseModel

All response models should extend `xAPI.ResponseModel`, which extends Pydantic's `BaseModel` with convenience methods and optional API metadata.

```python
class Coin(xAPI.ResponseModel):
    id: str
    symbol: str
    name: str
    market_cap: float | None = None
    current_price: float | None = None
```

**Convenience methods:**

```python
coin = await client.coins.coin(parameters=CoinParams.Bitcoin)

# Convert to dict (excludes unset fields by default)
coin.toDict()

# Convert to formatted JSON string
coin.toJson(indent=2)

# Access API metadata (method, path, elapsed time)
print(coin.api.endpoint)  # "GET coins/bitcoin in 0.35s"
```

**List responses:** When an API returns a JSON array instead of an object, use a `list[Model]` response type on the endpoint:

```python
class Category(xAPI.ResponseModel):
    id: str
    name: str

CategoryList = list[Category]

endpoint = xAPI.Endpoint(
    name="categories",
    path=CategoriesPath(),
    response=CategoryList,
)

categories = await client.coins.categories()  # returns list[Category]
```

---

## Authentication

xAPI supports API key authentication with three scoping levels:

```python
# Apply auth to ALL endpoints
auth = xAPI.APIKey(
    keyName="x-api-key",
    apiKey="your-secret-key",
    scope=xAPI.Scope.All
)

# Apply auth only to endpoints on resources with requireAuth=True
auth = xAPI.APIKey(
    keyName="x-api-key",
    apiKey="your-secret-key",
    scope=xAPI.Scope.Endpoint
)

# Disable auth
auth = xAPI.APIKey(
    keyName="x-api-key",
    apiKey="your-secret-key",
    scope=xAPI.Scope.Disabled
)
```

When `scope=Scope.All`, the auth key-value pair is added to both request headers and query parameters on every request.

When `scope=Scope.Endpoint`, auth is only applied to requests made through resources that have `requireAuth=True`.

---

## Rate Limiting

The built-in sliding window rate limiter prevents exceeding API quotas:

```python
rate_limiter = xAPI.RateLimiter(
    maxCalls=30,     # maximum number of calls
    perSecond=60,    # within this time window (seconds)
)

client = xAPI.Client(
    url="https://api.example.com/v1/",
    rateLimit=rate_limiter,
)
```

The rate limiter uses an async lock and automatically pauses requests when the limit is reached.

---

## Retry & Error Handling

### Retry Configuration

Retries use exponential backoff and only trigger on retriable errors (5xx, timeouts, connection errors). 4xx errors are raised immediately.

```python
retry = xAPI.RetryConfig(
    attempts=3,        # max retry attempts (default: 3)
    baseDelay=0.3,     # initial delay in seconds (default: 0.3)
    maxDelay=5.0,      # maximum delay in seconds (default: 5.0)
)
```

### Exception Hierarchy

xAPI provides specific exception types for different failure modes:

| Exception | When |
|---|---|
| `APIStatusError` | Any 4xx or 5xx response |
| `BadRequestError` | HTTP 400 |
| `AuthenticationError` | HTTP 401 |
| `PermissionDeniedError` | HTTP 403 |
| `NotFoundError` | HTTP 404 |
| `ConflictError` | HTTP 409 |
| `UnprocessableEntityError` | HTTP 422 |
| `RateLimitError` | HTTP 429 |
| `InternalServerError` | HTTP 5xx |
| `APITimeoutError` | Request timed out |
| `APIConnectionError` | Connection failed |
| `APIResponseValidationError` | Response doesn't match the Pydantic model |

```python
import xAPI

try:
    coin = await client.coins.coin(parameters=CoinParams.Bitcoin)
except xAPI.NotFoundError:
    print("Coin not found")
except xAPI.RateLimitError:
    print("Rate limited - slow down")
except xAPI.APIStatusError as e:
    print(f"API error {e.status_code}: {e.message}")
except xAPI.APIConnectionError:
    print("Could not connect to API")
except xAPI.APITimeoutError:
    print("Request timed out")
```

---

## Nested Data Unwrapping

Many APIs wrap their response in a `{"data": {...}}` envelope. xAPI automatically unwraps this by default, so your models only need to define the inner data structure.

```python
# API returns: {"data": {"total_market_cap": 2.5e12, "total_volume": 1e11}}
# Your model only needs:
class MarketData(xAPI.ResponseModel):
    total_market_cap: float
    total_volume: float
```

To disable this behavior, set `client.unsetNestedData = False`.

---

## Options (Enum Helpers)

xAPI provides base enum classes for defining typed option values:

```python
from xAPI import Options, IntOptions

# String-based options
Status = Options("Status", ["active", "inactive"])
Interval = Options("Interval", ["5m", "hourly", "daily"])

# Integer-based options
Days = IntOptions("Days", [("one", 1), ("seven", 7), ("thirty", 30)])
```

---

## Debug Logging

Enable debug logging to see detailed request/response information:

```python
client = xAPI.Client(
    url="https://api.example.com/v1/",
    debug=True,
)
```

This enables color-coded, structured logging for:
- Client operations (resource binding)
- HTTP requests (method, path, timing)
- Endpoint resolution
- Retry attempts

---

## Full Example

Here's a complete example using the CoinGecko API:

```python
import asyncio
import xAPI

# --- Response Models ---

class Coin(xAPI.ResponseModel):
    id: str
    symbol: str
    name: str
    description: dict | None = None
    market_data: dict | None = None

class CoinTickers(xAPI.ResponseModel):
    name: str
    tickers: list | None = None

class Category(xAPI.ResponseModel):
    id: str
    name: str

# --- Path Parameters ---

class CoinParams(xAPI.Parameters):
    Bitcoin = "bitcoin"
    Ethereum = "ethereum"
    Solana = "solana"

# --- Paths ---

class CoinByID(xAPI.Path[CoinParams]):
    endpointPath: str = "{id}"

class CoinTickersPath(xAPI.Path[CoinParams]):
    endpointPath: str = "{id}/tickers"

class CategoriesPath(xAPI.Path):
    endpointPath: str = "categories"

# --- Main ---

async def main():
    auth = xAPI.APIKey(
        keyName="x_cg_demo_api_key",
        apiKey="YOUR_KEY",
        scope=xAPI.Scope.All
    )

    async with xAPI.Client(
        url="https://api.coingecko.com/api/v3/",
        authentication=auth,
        rateLimit=xAPI.RateLimiter(maxCalls=30, perSecond=60),
        retry=xAPI.RetryConfig(attempts=3),
        debug=True
    ) as client:

        # Build the resource tree
        coins = xAPI.Resource("coins")
        coins.addEndpoints([
            xAPI.Endpoint(name="coin", path=CoinByID(), response=Coin),
            xAPI.Endpoint(name="tickers", path=CoinTickersPath(), response=CoinTickers),
            xAPI.Endpoint(name="categories", path=CategoriesPath(), response=list[Category]),
        ])
        client.add(coins)

        # Fetch a coin with query parameters
        query = xAPI.Query({
            "localization": False,
            "tickers": False,
            "market_data": False,
            "community_data": False,
            "developer_data": False,
            "sparkline": False,
        })

        try:
            bitcoin = await client.coins.coin(
                parameters=CoinParams.Bitcoin,
                query=query
            )
            print(f"{bitcoin.name} ({bitcoin.symbol})")
            print(bitcoin.toJson(indent=2))

            # Fetch categories (list response)
            categories = await client.coins.categories()
            for cat in categories[:5]:
                print(f"  - {cat.name}")

        except xAPI.NotFoundError:
            print("Resource not found")
        except xAPI.RateLimitError:
            print("Rate limited")
        except xAPI.APIStatusError as e:
            print(f"API error: {e.status_code}")

asyncio.run(main())
```

---

## API Reference

### `xAPI.Client(url, authentication?, timeout?, rateLimit?, retry?, headers?, debug?)`

The async HTTP client. Manages connections, auth, and the resource tree.

### `xAPI.Resource(name, prefix?, pathProtection?, requireAuth?)`

A named group of endpoints. Add to client with `client.add(resource)`.

### `xAPI.Endpoint(name, path, response?, method?, nameOverride?, strict?)`

A single API endpoint definition.

### `xAPI.Path[P]`

Protocol for URL path templates. Subclass and set `endpointPath`.

### `xAPI.Parameters`

Base `StrEnum` for typed path parameters.

### `xAPI.Query(queries)`

Query parameter builder. Filters out `"NOT_GIVEN"` values.

### `xAPI.APIKey(keyName, apiKey, scope)`

API key authentication with configurable scope.

### `xAPI.RateLimiter(maxCalls, perSecond)`

Sliding window rate limiter.

### `xAPI.RetryConfig(attempts?, baseDelay?, maxDelay?)`

Exponential backoff retry configuration.

### `xAPI.ResponseModel`

Base model for API responses. Extends Pydantic `BaseModel` with `toDict()` and `toJson()`.

---
## 📚 ・ xDev Utilities
This library is part of **xDev Utilities**. As set of power tool to streamline your workflow.

- **[xAPI](https://github.com/rkohl/xAPI)**: A lightweight, flexible asynchronous API client for Python built on Pydantic and httpx
- **[xEvents](https://github.com/rkohl/xEvents)**: A lightweight, thread-safe event system for Python

---
## License

BSD-3-Clause