xAPI

A lightweight, flexible asynchronous API client for Python built on httpx and pydantic.

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.

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

---

Features

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

$ pip install xAPI-client

Requirements: Python 3.12+

---

Quick Start

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.

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.

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:

# 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:

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.

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:

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

Call an endpoint:

# 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.

# 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.

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.

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

Convenience methods:

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:

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:

# 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:

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.

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

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.

# 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:

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:

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:

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: A lightweight, flexible asynchronous API client for Python built on Pydantic and httpx
• xEvents: A lightweight, thread-safe event system for Python

---

License

BSD-3-Clause