API Reference

This page provides detailed API documentation for using apiout programmatically.

Modules

apiout.cli

apiout.fetcher

API fetching and client management for apiout.

This module provides functionality for: - Fetching data from API endpoints defined in TOML configurations - Managing shared client instances across multiple API calls - Processing post-processors that combine multiple API results - Serializing API responses according to configuration

apiout.fetcher.resolve_serializer(api_config: dict[str, Any], global_serializers: dict[str, Any] | None = None, client_ref: str | None = None) → dict[str, Any][source]

Resolve serializer configuration from API config with client-scoped namespace support.

Resolution order: 1. Inline dict (api_config[“serializer”] is dict) - highest priority 2. Explicit dotted reference (e.g., “client.serializer_name”) 3. Client-scoped lookup (e.g., serializers.{client_ref}.{name}) 4. Global lookup (e.g., serializers.{name}) 5. Empty dict (no serializer found)

Args:

api_config: API configuration dict containing optional ‘serializer’ key global_serializers: Optional dict of named serializer configurations client_ref: Optional client reference name for scoped serializer lookup

Returns:

Resolved serializer configuration dict, or empty dict if none found

Examples:

>>> # Global serializer
>>> api_config = {"serializer": "my_serializer"}
>>> global_serializers = {"my_serializer": {"fields": {"name": "name"}}}
>>> resolve_serializer(api_config, global_serializers)
{'fields': {'name': 'name'}}

>>> # Client-scoped serializer
>>> api_config = {"serializer": "data", "client": "btc_price"}
>>> global_serializers = {"btc_price.data": {"fields": {"value": "usd"}}}
>>> resolve_serializer(api_config, global_serializers, client_ref="btc_price")
{'fields': {'value': 'usd'}}

>>> # Explicit dotted reference
>>> api_config = {"serializer": "btc_price.data"}
>>> global_serializers = {"btc_price.data": {"fields": {"value": "usd"}}}
>>> resolve_serializer(api_config, global_serializers)
{'fields': {'value': 'usd'}}

apiout.fetcher.fetch_api_data(api_config: dict[str, Any], global_serializers: dict[str, Any] | None = None, shared_clients: dict[str, Any] | None = None, client_configs: dict[str, Any] | None = None, user_params: dict[str, str] | None = None) → Any[source]

Fetch data from an API endpoint based on configuration.

Dynamically imports a module, instantiates or reuses a client class, and calls the specified method. Supports shared client instances when using client references.

Args:

api_config: API configuration dict with keys:

module: Python module to import (required)
method: Method name to call on client (required)
client: Reference to a client config name (optional)
client_class: Class name to instantiate (default: “Client”)
init_params: Params for client initialization (optional)
url: URL parameter to pass to method (optional)
params: Additional parameters for method (optional)

method_params: Default values for method parameters (optional)
- serializer: Serializer config or reference (optional)

global_serializers: Named serializer configurations shared_clients: Dict to store/retrieve shared client instances client_configs: Dict of named client configurations user_params: Dict of user-provided runtime parameters

Returns:

Serialized API response data, or error dict if fetch failed

Example:

>>> api_config = {
...     "module": "requests",
...     "client_class": "Session",
...     "method": "get",
...     "url": "https://api.example.com/data"
... }
>>> result = fetch_api_data(api_config)

apiout.fetcher.process_post_processor(post_processor_config: dict[str, Any], api_results: dict[str, Any], global_serializers: dict[str, Any] | None = None) → Any[source]

Process data from multiple APIs using a post-processor class.

Post-processors combine results from multiple API calls by instantiating a class with the API results as arguments, or calling a method on an instance with the results.

Args:

post_processor_config: Post-processor configuration dict with keys:

module: Python module to import (required)
class: Class name to instantiate (required)
inputs: List of API result names to pass as args (required)
method: Method name to call on instance (optional)
serializer: Serializer config or reference (optional)

api_results: Dict of API results by name global_serializers: Named serializer configurations

Returns:

Serialized post-processor result, or error dict if processing failed

Example:

>>> post_processor_config = {
...     "module": "mymodule",
...     "class": "DataCombiner",
...     "inputs": ["api1", "api2"]
... }
>>> api_results = {"api1": {"value": 1}, "api2": {"value": 2}}
>>> result = process_post_processor(post_processor_config, api_results)

class apiout.fetcher.ApiClient(config_paths: str | Path | list[str | Path], user_params: dict[str, str] | None = None)[source]

Bases: object

Stateful API client with configuration management and result caching.

ApiClient provides a high-level interface for loading API configurations from TOML files, fetching data from multiple APIs with shared client instances, and caching results for repeated access without re-fetching.

Supports: - Loading single or multiple TOML configuration files - Automatic merging of APIs, serializers, and post-processors - Shared client instance management via client references - Result caching with success/failure tracking - Timestamp tracking for each API call

Attributes:

config_paths: List of loaded configuration file paths apis: List of API configurations from all loaded files serializers: Dict of named serializer configurations post_processors: List of post-processor configurations clients: Dict of named client configurations shared_clients: Dict of shared client instances by reference name results: Dict of API results by name (cached after fetch) status: Dict of status info by name (success, error, timestamp) last_fetch_time: Timestamp of the most recent fetch() call

Example:

>>> # Single config file
>>> client = ApiClient("config.toml")
>>> results = client.fetch()
>>> cached = client.get_results()
>>>
>>> # Multiple config files
>>> client = ApiClient(["api_config.toml", "serializers.toml"])
>>> results = client.fetch()
>>> status = client.get_status()
>>> successful = client.get_successful_results()

__init__(config_paths: str | Path | list[str | Path], user_params: dict[str, str] | None = None)[source]

Initialize ApiClient with one or more configuration files.

Args:

config_paths: Single path or list of paths to TOML configuration files.: All configs are loaded and merged during initialization.

user_params: Optional dict of user-provided runtime parameters

fetch() → dict[str, Any][source]

Fetch data from all configured APIs and post-processors.

Executes all API calls using shared client instances where configured, then runs post-processors on the results. Updates results, status, and last_fetch_time attributes.

Returns:

Dict mapping API/post-processor names to their results

Example:

>>> client = ApiClient("config.toml")
>>> results = client.fetch()
>>> print(results["my_api"])
{'data': 'value'}

get_results() → dict[str, Any][source]

Get cached results without re-fetching.

Returns:

Dict of cached results from the last fetch() call

Example:

>>> client = ApiClient("config.toml")
>>> client.fetch()
>>> cached = client.get_results()  # No network call

get_status() → dict[str, dict][source]

Get status information for all APIs and post-processors.

Returns:

Dict mapping names to status dicts with keys: - success: bool indicating if fetch/processing succeeded - error: error message if failed, None otherwise - timestamp: Unix timestamp of the operation

Example:

>>> client = ApiClient("config.toml")
>>> client.fetch()
>>> status = client.get_status()
>>> print(status["my_api"])
{'success': True, 'error': None, 'timestamp': 1234567890.123}

get_successful_results() → dict[str, Any][source]

Get only results from successful API calls and post-processors.

Returns:

Dict containing only results where status[‘success’] is True

Example:

>>> client = ApiClient("config.toml")
>>> client.fetch()
>>> successful = client.get_successful_results()

apiout.serializer

apiout.serializer.serialize_key(key: Any) → str[source]

apiout.serializer.serialize_value(obj: Any) → Any[source]

apiout.serializer.call_method_or_attr(obj: Any, name: str) → Any[source]

apiout.serializer.traverse_path(obj: Any, path_parts: list[str], parse_json: bool = False) → Any[source]

apiout.serializer.apply_field_mapping(obj: Any, field_config: Any) → Any[source]

apiout.serializer.apply_config_serializer(responses: Any, serializer_config: dict[str, Any]) → Any[source]

apiout.serializer.serialize_response(responses: Any, serializer_config: dict[str, Any]) → Any[source]

apiout.generator

apiout.generator.is_simple_type(obj: Any) → bool[source]

apiout.generator.is_collection(obj: Any) → bool[source]

apiout.generator.get_methods_and_attrs(obj: Any) → tuple[list[str], list[str]][source]

apiout.generator.analyze_object(obj: Any, max_depth: int = 3, current_depth: int = 0, visited: set[int] | None = None) → dict[str, Any][source]

apiout.generator.generate_serializer_config(analysis: dict[str, Any], prefix: str = '') → dict[str, Any][source]

apiout.generator.generate_toml_serializer(name: str, fields: dict[str, Any], indent: int = 0) → str[source]

apiout.generator.introspect_and_generate(module_name: str, client_class: str, method_name: str, url: str | None = None, params: dict[str, Any] | None = None, init_params: dict[str, Any] | None = None, serializer_name: str = 'generated', method_params: dict[str, Any] | None = None) → str[source]

apiout.generator.generate_api_toml(name: str, module_name: str, client_class: str, method_name: str, client_ref: str | None = None, init_params: dict[str, Any] | None = None, url: str | None = None, params: dict[str, Any] | None = None, method_params: dict[str, Any] | None = None) → str[source]

apiout.generator.introspect_post_processor_and_generate(module_name: str, class_name: str, method_name: str, input_modules: list[dict[str, Any]], serializer_name: str = 'generated') → str[source]

Generate serializer for a post-processor by calling it with sample data.

Args:

module_name: Module containing the post-processor class class_name: Post-processor class name method_name: Optional method to call on the instance input_modules: List of dicts with ‘module’, ‘client_class’,

‘method’, optional ‘init_params’, ‘url’, ‘params’ keys

serializer_name: Name for the generated serializer

Functions

fetch_api_data

apiout.fetcher.fetch_api_data(api_config: dict[str, Any], global_serializers: dict[str, Any] | None = None, shared_clients: dict[str, Any] | None = None, client_configs: dict[str, Any] | None = None, user_params: dict[str, str] | None = None) → Any[source]

Fetch data from an API endpoint based on configuration.

Dynamically imports a module, instantiates or reuses a client class, and calls the specified method. Supports shared client instances when using client references.

Args:

api_config: API configuration dict with keys:

module: Python module to import (required)
method: Method name to call on client (required)
client: Reference to a client config name (optional)
client_class: Class name to instantiate (default: “Client”)
init_params: Params for client initialization (optional)
url: URL parameter to pass to method (optional)
params: Additional parameters for method (optional)

method_params: Default values for method parameters (optional)
- serializer: Serializer config or reference (optional)

global_serializers: Named serializer configurations shared_clients: Dict to store/retrieve shared client instances client_configs: Dict of named client configurations user_params: Dict of user-provided runtime parameters

Returns:

Serialized API response data, or error dict if fetch failed

Example:

>>> api_config = {
...     "module": "requests",
...     "client_class": "Session",
...     "method": "get",
...     "url": "https://api.example.com/data"
... }
>>> result = fetch_api_data(api_config)

Fetches data from an API using the provided configuration.

Parameters:

api_config (Dict[str, Any]): API configuration dictionary
global_serializers (Optional[Dict[str, Any]]): Global serializer configurations
user_params (Optional[Dict[str, str]]): Runtime parameters to override method_params

Returns:

Any: Serialized API response or error dictionary

Example:

from apiout.fetcher import fetch_api_data

config = {
    "name": "test",
    "module": "requests",
    "client_class": "Session",
    "method": "get",
    "url": "https://api.example.com",
    "params": {}
}

result = fetch_api_data(config)

resolve_serializer

apiout.fetcher.resolve_serializer(api_config: dict[str, Any], global_serializers: dict[str, Any] | None = None, client_ref: str | None = None) → dict[str, Any][source]

Resolve serializer configuration from API config with client-scoped namespace support.

Resolution order: 1. Inline dict (api_config[“serializer”] is dict) - highest priority 2. Explicit dotted reference (e.g., “client.serializer_name”) 3. Client-scoped lookup (e.g., serializers.{client_ref}.{name}) 4. Global lookup (e.g., serializers.{name}) 5. Empty dict (no serializer found)

Args:

api_config: API configuration dict containing optional ‘serializer’ key global_serializers: Optional dict of named serializer configurations client_ref: Optional client reference name for scoped serializer lookup

Returns:

Resolved serializer configuration dict, or empty dict if none found

Examples:

>>> # Global serializer
>>> api_config = {"serializer": "my_serializer"}
>>> global_serializers = {"my_serializer": {"fields": {"name": "name"}}}
>>> resolve_serializer(api_config, global_serializers)
{'fields': {'name': 'name'}}

>>> # Client-scoped serializer
>>> api_config = {"serializer": "data", "client": "btc_price"}
>>> global_serializers = {"btc_price.data": {"fields": {"value": "usd"}}}
>>> resolve_serializer(api_config, global_serializers, client_ref="btc_price")
{'fields': {'value': 'usd'}}

>>> # Explicit dotted reference
>>> api_config = {"serializer": "btc_price.data"}
>>> global_serializers = {"btc_price.data": {"fields": {"value": "usd"}}}
>>> resolve_serializer(api_config, global_serializers)
{'fields': {'value': 'usd'}}

Resolves serializer configuration from API config and global serializers.

Parameters:

api_config (Dict[str, Any]): API configuration dictionary
global_serializers (Optional[Dict[str, Any]]): Global serializer configurations

Returns:

Dict[str, Any]: Resolved serializer configuration

serialize_response

apiout.serializer.serialize_response(responses: Any, serializer_config: dict[str, Any]) → Any[source]

Serializes API response using the provided serializer configuration.

Parameters:

responses (Any): API response object(s)
serializer_config (Dict[str, Any]): Serializer configuration

Returns:

Any: Serialized response

serialize_value

apiout.serializer.serialize_value(obj: Any) → Any[source]

Converts a Python object to a JSON-serializable value.

Parameters:

obj (Any): Object to serialize

Returns:

Any: JSON-serializable value

apply_field_mapping

apiout.serializer.apply_field_mapping(obj: Any, field_config: Any) → Any[source]

Applies field mapping configuration to an object.

Parameters:

obj (Any): Object to process
field_config (Dict[str, Any]): Field mapping configuration

Returns:

Any: Mapped result

call_method_or_attr

apiout.serializer.call_method_or_attr(obj: Any, name: str) → Any[source]

Calls a method or accesses an attribute on an object.

Parameters:

obj (Any): Object to access
name (str): Method or attribute name

Returns:

Any: Result of method call or attribute value

introspect_and_generate

apiout.generator.introspect_and_generate(module_name: str, client_class: str, method_name: str, url: str | None = None, params: dict[str, Any] | None = None, init_params: dict[str, Any] | None = None, serializer_name: str = 'generated', method_params: dict[str, Any] | None = None) → str[source]

Introspects an API response and generates a TOML serializer configuration.

Parameters:

module (str): Python module name
client_class (str): Client class name
method (str): Method name to call
url (str): API URL
params (Dict[str, Any]): API parameters
name (str): Serializer name

Returns:

str: TOML serializer configuration

analyze_object

apiout.generator.analyze_object(obj: Any, max_depth: int = 3, current_depth: int = 0, visited: set[int] | None = None) → dict[str, Any][source]

Analyzes an object’s structure and returns metadata about its attributes and methods.

Parameters:

obj (Any): Object to analyze
depth (int): Maximum recursion depth
visited (Optional[Set[int]]): Set of visited object IDs

Returns:

Dict[str, Any]: Object analysis metadata

generate_serializer_config

apiout.generator.generate_serializer_config(analysis: dict[str, Any], prefix: str = '') → dict[str, Any][source]

Generates a serializer configuration dictionary from object analysis.

Parameters:

analysis (Dict[str, Any]): Object analysis metadata
name (str): Serializer name

Returns:

str: TOML serializer configuration

Classes

Configuration Structure

API Configuration

{: “name”: str, # Required: API identifier “module”: str, # Required: Python module name “client_class”: str, # Optional: Client class name (default: “Client”) “method”: str, # Required: Method name to call “url”: str, # Required: API URL “serializer”: str | dict, # Optional: Serializer reference or config “params”: dict, # Optional: API parameters “method_params”: dict # Optional: Default method parameters

}

Serializer Configuration

{
    "fields": {
        "output_field": str,  # Simple attribute mapping
        "nested_field": {
            "method": str,    # Method to call
            "fields": dict    # Nested field mappings
        },
        "collection": {
            "iterate": {
                "count": str,      # Method returning count
                "item": str,       # Method taking index
                "fields": dict     # Fields for each item
            }
        }
    }
}

Error Handling

Common Error Types

Import Errors

{
    "error": "Failed to import module: ModuleNotFoundError: No module named 'xxx'"
}

Attribute Errors

{
    "error": "Failed to access class or method: 'module' object has no attribute 'Class'"
}

API Call Errors

{
    "error": "Failed to fetch data: <exception details>"
}

Usage Examples

Programmatic Usage

Basic Fetch

from apiout.fetcher import fetch_api_data

api_config = {
    "name": "weather",
    "module": "openmeteo_requests",
    "client_class": "Client",
    "method": "weather_api",
    "url": "https://api.open-meteo.com/v1/forecast",
    "params": {
        "latitude": 52.52,
        "longitude": 13.41,
        "current": ["temperature_2m"]
    }
}

result = fetch_api_data(api_config)
print(result)

With Serializer

from apiout.fetcher import fetch_api_data

api_config = {
    "name": "weather",
    "module": "openmeteo_requests",
    "method": "weather_api",
    "url": "https://api.open-meteo.com/v1/forecast",
    "serializer": "openmeteo",
    "params": {"latitude": 52.52, "longitude": 13.41}
}

global_serializers = {
    "openmeteo": {
        "fields": {
            "latitude": "Latitude",
            "longitude": "Longitude",
            "current": {
                "method": "Current",
                "fields": {"time": "Time"}
            }
        }
    }
}

result = fetch_api_data(api_config, global_serializers)
print(result)

Custom Serialization

from apiout.serializer import serialize_response, apply_field_mapping

# Assuming you have a response object
response = api_call()

# Define field mapping
field_config = {
    "id": "Id",
    "name": "Name",
    "data": {
        "method": "GetData",
        "fields": {
            "value": "Value",
            "timestamp": "Timestamp"
        }
    }
}

# Apply mapping
result = apply_field_mapping(response, field_config)
print(result)

Generate Configuration

from apiout.generator import introspect_and_generate

config = introspect_and_generate(
    module="openmeteo_requests",
    client_class="Client",
    method="weather_api",
    url="https://api.open-meteo.com/v1/forecast",
    params={"latitude": 52.52, "longitude": 13.41},
    name="openmeteo"
)

print(config)  # TOML serializer configuration