BiomapperClient API Reference

The BiomapperClient class provides the main interface for interacting with the BioMapper API. It supports both synchronous and asynchronous operation modes.

class BiomapperClient(base_url='http://localhost:8000', api_key=None, timeout=300, auto_retry=True, max_retries=3)

Enhanced Biomapper API client for strategy execution.

Parameters:

base_url (str) – Base URL of the BioMapper API server
api_key (str) – Optional API key for authentication
timeout (int) – Request timeout in seconds (default: 300)
auto_retry (bool) – Whether to automatically retry failed requests
max_retries (int) – Maximum number of retries

Example:

from biomapper.client import BiomapperClient

# Create client instance
client = BiomapperClient("http://localhost:8000")

# Execute a strategy
result = client.run("protein_harmonization")

Synchronous Methods

run(strategy, parameters=None, context=None, wait=True, watch=False)

Execute a strategy synchronously (recommended for most users).

Parameters:

strategy (Union[str, Path, dict]) – Name of the strategy, path to YAML file, or inline strategy dict
parameters (dict) – Optional parameter overrides
context (Union[dict, ExecutionContext]) – Optional execution context (defaults to empty context)
wait (bool) – Whether to wait for completion (default: True)
watch (bool) – Display real-time progress (default: False)

Returns:

Job object if wait=False, StrategyResult if wait=True

Return type:

Union[Job, StrategyResult]

Example:

result = client.run("metabolomics_baseline", parameters={
    "input_file": "/data/metabolites.csv",
    "threshold": 0.95
})

Asynchronous Methods

async execute_strategy(strategy, parameters=None, context=None, options=None)

Execute a strategy asynchronously.

Parameters:

strategy (Union[str, Path, dict]) – Name of the strategy, path to YAML file, or inline strategy dict
parameters (dict) – Optional parameter overrides
context (Union[dict, ExecutionContext]) – Optional execution context (defaults to empty context)
options (ExecutionOptions) – Execution options

Returns:

Job object with execution details

Return type:

Job

async wait_for_job(job_id, poll_interval=1.0)

Wait for a job to complete.

Parameters:

job_id (str) – Job identifier
poll_interval (float) – Seconds between status checks

Returns:

Final job result

Return type:

StrategyResult

Required context structure:

context = {
    "current_identifiers": [],     # List of active identifiers
    "datasets": {},                 # Named datasets
    "statistics": {},               # Accumulated statistics
    "output_files": [],            # Generated file paths
    "metadata": {}                 # Optional metadata
}

Example:

import asyncio

async def run_strategy():
    async with BiomapperClient() as client:
        context = {
            "current_identifiers": [],
            "datasets": {"input": [...]},
            "statistics": {},
            "output_files": []
        }
        result = await client.execute_strategy("protein_harmonization", context)
        return result

result = asyncio.run(run_strategy())

Context Manager Support

The client supports both synchronous and asynchronous context managers:

Asynchronous Context Manager:

async with BiomapperClient() as client:
    result = await client.execute_strategy("strategy_name", context)

Synchronous Usage (no context manager needed):

client = BiomapperClient()
result = client.run("strategy_name")

Exception Classes

class ApiError: Raised when the API returns a non-success status code.

class NetworkError: Raised for network-related issues (connection, timeout).

class ValidationError: Raised when request validation fails.

class StrategyNotFoundError: Raised when a requested strategy doesn’t exist.

class JobNotFoundError: Raised when a job ID is not found.

class TimeoutError: Raised when operation exceeds timeout.

class FileUploadError: Raised when file upload fails.

Utility Functions

The biomapper_client package also provides utility functions for CLI usage:

run_strategy(strategy_path, output_dir=None, verbose=False)

Execute a strategy from the command line.

Parameters:

strategy_path (str) – Path to strategy YAML file
output_dir (str) – Optional output directory
verbose (bool) – Enable verbose output

Returns:

Execution results

Return type:

dict

parse_parameters(param_strings)

Parse command-line parameter strings into a dictionary.

Parameters:: param_strings (list) – List of “key=value” strings
Returns:: Parameter dictionary
Return type:: dict

Example:

params = parse_parameters([
    "input_file=/data/proteins.csv",
    "threshold=0.95",
    "output_dir=/results"
])
# Returns: {"input_file": "/data/proteins.csv", "threshold": 0.95, "output_dir": "/results"}

Response Schemas

Successful Response:

{
    "status": "success",
    "results": {
        "datasets": {
            "dataset_name": [...]  # Named datasets from workflow
        },
        "statistics": {
            "total_records": int,
            "processing_time": float,
            # Additional action-specific statistics
        },
        "output_files": [
            # List of generated file paths
        ],
        "metadata": {
            # Strategy metadata and parameters
        }
    },
    "execution_time": float  # Total execution time in seconds
}

Error Response:

{
    "status": "error",
    "detail": "Error message",
    "error_type": "ErrorClassName",
    "traceback": "..."  # Optional stack trace
}

Configuration

The client can be configured through environment variables:

Environment Variable	Description
`BIOMAPPER_API_URL`	Override default API URL
`BIOMAPPER_API_KEY`	API key for authentication
`BIOMAPPER_TIMEOUT`	Request timeout in seconds
`BIOMAPPER_MAX_RETRIES`	Maximum retry attempts
`BIOMAPPER_AUTO_RETRY`	Enable/disable auto-retry (true/false)

Thread Safety

The synchronous run() method is thread-safe
The async client should use separate instances per thread
Context managers handle resource cleanup automatically

Performance Considerations

Default timeout: 3 hours (for large datasets)
Automatic retry on network errors (configurable)
Connection pooling for multiple requests
Chunked uploads for large files (>10MB)

Version Compatibility

Client version: 0.5.2
Compatible API versions: 0.5.2+
Python: 3.11+
Dependencies: httpx, pydantic 2.0+

BiomapperClient API Reference

Synchronous Methods

Asynchronous Methods

Context Manager Support

Exception Classes

Utility Functions

Response Schemas

Configuration

Thread Safety

Performance Considerations

Version Compatibility

See Also

Verification Sources