apilinker

ApiLinker

A universal bridge to connect, map, and automate data transfer between any two REST APIs

ApiLinker is an open-source Python package that simplifies the integration of REST APIs by providing a universal bridging solution. Built for developers, data engineers, and researchers who need to connect different systems without writing repetitive boilerplate code.

🌟 Features

• 🔄 Universal Connectivity - Connect any two REST APIs with simple configuration
• 🗺️ Powerful Mapping - Transform data between APIs with field mapping and path expressions
• 📊 Data Transformation - Apply built-in or custom transformations to your data
• 🔒 Comprehensive Authentication - Support for API Key, Bearer Token, Basic Auth, and OAuth2
• 📝 Flexible Configuration - Use YAML/JSON or configure programmatically in Python
• 🕒 Automated Scheduling - Run syncs once, on intervals, or using cron expressions
• 📋 Data Validation - Validate data with schemas and custom rules
• 🔌 Plugin Architecture - Extend with custom connectors, transformers, and authentication methods
• 📈 Pagination Handling - Automatic handling of paginated API responses
• 🔍 Error Recovery - Built-in retry logic and error handling
• 📦 Minimal Dependencies - Lightweight core with minimal external requirements

📋 Table of Contents

• Installation
• Quick Start
• Configuration
• Authentication Methods
• Field Mapping
• Data Transformations
• Scheduling
• Command Line Interface
• Python API
• Examples
• Extending ApiLinker
• Contributing
• Documentation
• License

🚀 Installation

Standard Installation

Install ApiLinker using pip (Python's package manager):

pip install apilinker

If you're using Windows, you might need to use:

py -m pip install apilinker

Make sure you have Python 3.8 or newer installed. To check your Python version:

python --version
# or
py --version

Development Installation

To install from source (for contributing or customizing):

# Clone the repository
git clone https://github.com/kkartas/apilinker.git
cd apilinker

# Install in development mode with dev dependencies
pip install -e ".[dev]"

# Install with documentation tools
pip install -e ".[docs]"

Verifying Installation

To verify ApiLinker is correctly installed, run:

python -c "import apilinker; print(apilinker.__version__)"

You should see the version number printed if installation was successful.

🎯 Beginner's Guide

New to API integration? Follow this step-by-step guide to get started with ApiLinker.

Step 1: Install ApiLinker

pip install apilinker

Step 2: Create Your First API Connection

Let's connect to a public API (Weather API) and print some data:

from apilinker import ApiLinker

# Create an API connection
linker = ApiLinker()

# Configure a simple source
linker.add_source(
    type="rest",
    base_url="https://api.openweathermap.org/data/2.5",
    endpoints={
        "get_weather": {
            "path": "/weather",
            "method": "GET",
            "params": {
                "q": "London",
                "appid": "YOUR_API_KEY"  # Get a free key at openweathermap.org
            }
        }
    }
)

# Fetch data from the API
weather_data = linker.fetch("get_weather")

# Print results
print(f"Temperature: {weather_data['main']['temp']} K")
print(f"Conditions: {weather_data['weather'][0]['description']}")

Step 3: Save the Script and Run It

Save the above code as weather.py and run it:

python weather.py

Step 4: Try a Data Transformation

Let's convert the temperature from Kelvin to Celsius:

# Add this to your script
def kelvin_to_celsius(kelvin_value):
    return kelvin_value - 273.15

linker.mapper.register_transformer("kelvin_to_celsius", kelvin_to_celsius)

# Get the temperature in Celsius
temp_kelvin = weather_data['main']['temp']
temp_celsius = linker.mapper.transform(temp_kelvin, "kelvin_to_celsius")

print(f"Temperature: {temp_celsius:.1f}°C")

Common Beginner Issues

• ImportError: Make sure ApiLinker is installed (pip install apilinker)
• API Key errors: Register for a free API key at the service you're using
• Connection errors: Check your internet connection and API endpoint URL
• TypeError: Make sure you're passing the correct data types to functions

🏁 Quick Start

Using the CLI

Create a configuration file config.yaml:

source:
  type: rest
  base_url: https://api.example.com/v1
  auth:
    type: bearer
    token: ${SOURCE_API_TOKEN}  # Reference environment variable
  endpoints:
    list_items:
      path: /items
      method: GET
      params:
        updated_since: "{{last_sync}}"  # Template variable
      pagination:
        data_path: data
        next_page_path: meta.next_page
        page_param: page
target:
  type: rest
  base_url: https://api.destination.com/v2
  auth:
    type: api_key
    header: X-API-Key
    key: ${TARGET_API_KEY}
  endpoints:
    create_item:
      path: /items
      method: POST

mapping:
  - source: list_items
    target: create_item
    fields:
      - source: id
        target: external_id
      - source: name
        target: title
      - source: description
        target: body.content
      - source: created_at
        target: metadata.created
        transform: iso_to_timestamp
      # Conditional field mapping
      - source: tags
        target: labels
        condition:
          field: tags
          operator: exists
        transform: lowercase

schedule:
  type: interval
  minutes: 60

logging:
  level: INFO
  file: apilinker.log

Run a sync with:

apilinker sync --config config.yaml

Run a dry run to see what would happen without making changes:

apilinker sync --config config.yaml --dry-run

Run a scheduled sync based on the configuration:

apilinker run --config config.yaml

Using as a Python Library

from apilinker import ApiLinker

# Initialize with config file
linker = ApiLinker(config_path="config.yaml")

# Or configure programmatically
linker = ApiLinker()

# Step 1: Set up your source API connection
linker.add_source(
    type="rest",                          # API type (REST is most common)
    base_url="https://api.github.com",   # Base URL of the API
    auth={                              # Authentication details
        "type": "bearer",              # Using bearer token authentication
        "token": "${GITHUB_TOKEN}"     # Reference to an environment variable
    },
    endpoints={                          # Define API endpoints
        "list_issues": {               # A name you choose for this endpoint
            "path": "/repos/owner/repo/issues",  # API path
            "method": "GET",           # HTTP method
            "params": {"state": "all"}  # Query parameters
        }
    }
)

# Step 2: Set up your target API connection
linker.add_target(
    type="rest",
    base_url="https://gitlab.com/api/v4",
    auth={
        "type": "bearer",
        "token": "${GITLAB_TOKEN}"
    },
    endpoints={
        "create_issue": {
            "path": "/projects/123/issues",
            "method": "POST"           # This endpoint will receive data
        }
    }
)

# Step 3: Define how data maps from source to target
linker.add_mapping(
    source="list_issues",               # Source endpoint name (from Step 1)
    target="create_issue",              # Target endpoint name (from Step 2)
    fields=[                            # Field mapping instructions
        {"source": "title", "target": "title"},           # Map source title → target title
        {"source": "body", "target": "description"}      # Map source body → target description
    ]
)

# Step 4: Execute the sync (one-time)
result = linker.sync()
print(f"Synced {result.count} records")

# Step 5 (Optional): Set up scheduled syncing
linker.add_schedule(interval_minutes=60)  # Run every hour
linker.start_scheduled_sync()

Step-by-Step Explanation:

• Import the library: from apilinker import ApiLinker
• Create an instance: linker = ApiLinker()
• Configure source API: Define where to get data from
• Configure target API: Define where to send data to
• Create mappings: Define how fields translate between APIs
• Run the sync: Either once or on a schedule

🔧 Configuration

ApiLinker uses a YAML configuration format with these main sections:

Source and Target API Configuration

Both source and target sections follow the same format:

source:  # or target:
  type: rest  # API type
  base_url: https://api.example.com/v1  # Base URL
  auth:  # Authentication details
    # ...
  endpoints:  # API endpoints
    # ...
  timeout: 30  # Request timeout in seconds (optional)
  retry_count: 3  # Number of retries (optional)

Authentication Methods

ApiLinker supports multiple authentication methods:

# API Key Authentication
auth:
  type: api_key
  key: your_api_key  # Or ${API_KEY_ENV_VAR}
  header: X-API-Key  # Header name

# Bearer Token Authentication
auth:
  type: bearer
  token: your_token  # Or ${TOKEN_ENV_VAR}

# Basic Authentication
auth:
  type: basic
  username: your_username  # Or ${USERNAME_ENV_VAR}
  password: your_password  # Or ${PASSWORD_ENV_VAR}

# OAuth2 Client Credentials
auth:
  type: oauth2_client_credentials
  client_id: your_client_id  # Or ${CLIENT_ID_ENV_VAR}
  client_secret: your_client_secret  # Or ${CLIENT_SECRET_ENV_VAR}
  token_url: https://auth.example.com/token
  scope: read write  # Optional

Field Mapping

Mappings define how data is transformed between source and target:

mapping:
  - source: source_endpoint_name
    target: target_endpoint_name
    fields:
      # Simple field mapping
      - source: id
        target: external_id
      
      # Nested field mapping
      - source: user.profile.name
        target: user_name
      
      # With transformation
      - source: created_at
        target: timestamp
        transform: iso_to_timestamp
      
      # Multiple transformations
      - source: description
        target: summary
        transform:
          - strip
          - lowercase
      
      # Conditional mapping
      - source: status
        target: active_status
        condition:
          field: status
          operator: eq  # eq, ne, exists, not_exists, gt, lt
          value: active

🔄 Data Transformations

ApiLinker provides built-in transformers for common operations:

Transformer	Description
iso_to_timestamp	Convert ISO date to Unix timestamp
timestamp_to_iso	Convert Unix timestamp to ISO date
lowercase	Convert string to lowercase
uppercase	Convert string to uppercase
strip	Remove whitespace from start/end
to_string	Convert value to string
to_int	Convert value to integer
to_float	Convert value to float
to_bool	Convert value to boolean
default_empty_string	Return empty string if null
default_zero	Return 0 if null
none_if_empty	Return null if empty string

You can also create custom transformers:

def phone_formatter(value):
    """Format phone numbers to E.164 format."""
    if not value:
        return None
    digits = re.sub(r'\D', '', value)
    if len(digits) == 10:
        return f"+1{digits}"
    return f"+{digits}"

# Register with ApiLinker
linker.mapper.register_transformer("phone_formatter", phone_formatter)

📊 Examples

GitHub to GitLab Issue Migration

from apilinker import ApiLinker

# Configure ApiLinker
linker = ApiLinker(
    source_config={
        "type": "rest",
        "base_url": "https://api.github.com",
        "auth": {"type": "bearer", "token": github_token},
        "endpoints": {
            "list_issues": {
                "path": f"/repos/{owner}/{repo}/issues",
                "method": "GET",
                "params": {"state": "all"},
                "headers": {"Accept": "application/vnd.github.v3+json"}
            }
        }
    },
    target_config={
        "type": "rest",
        "base_url": "https://gitlab.com/api/v4",
        "auth": {"type": "bearer", "token": gitlab_token},
        "endpoints": {
            "create_issue": {
                "path": f"/projects/{project_id}/issues",
                "method": "POST"
            }
        }
    }
)

# Custom transformer for labels
linker.mapper.register_transformer(
    "github_labels_to_gitlab",
    lambda labels: [label["name"] for label in labels] if labels else []
)

# Add mapping
linker.add_mapping(
    source="list_issues",
    target="create_issue",
    fields=[
        {"source": "title", "target": "title"},
        {"source": "body", "target": "description"},
        {"source": "labels", "target": "labels", "transform": "github_labels_to_gitlab"},
        {"source": "state", "target": "state"}
    ]
)

# Run the migration
result = linker.sync()
print(f"Migrated {result.count} issues from GitHub to GitLab")

More Examples

See the examples directory for more use cases:

• Salesforce to HubSpot contact sync
• CSV file to REST API import
• Weather API data collection
• Custom plugin development

💻 Common Use Cases with Examples

1. Sync Data Between Two APIs

This example shows how to sync customer data from CRM to a marketing platform:

from apilinker import ApiLinker
import os

# Set environment variables securely before running
# os.environ["CRM_API_KEY"] = "your_crm_api_key"
# os.environ["MARKETING_API_KEY"] = "your_marketing_api_key"

# Initialize ApiLinker
linker = ApiLinker()

# Configure CRM source
linker.add_source(
    type="rest",
    base_url="https://api.crm-platform.com/v2",
    auth={
        "type": "api_key",
        "header": "X-API-Key",
        "key": "${CRM_API_KEY}"  # Uses environment variable
    },
    endpoints={
        "get_customers": {
            "path": "/customers",
            "method": "GET",
            "params": {"last_modified_after": "2023-01-01"}
        }
    }
)

# Configure marketing platform target
linker.add_target(
    type="rest",
    base_url="https://api.marketing-platform.com/v1",
    auth={
        "type": "api_key",
        "header": "Authorization", 
        "key": "${MARKETING_API_KEY}"  # Uses environment variable
    },
    endpoints={
        "create_contact": {
            "path": "/contacts",
            "method": "POST"
        }
    }
)

# Define field mapping with transformations
linker.add_mapping(
    source="get_customers",
    target="create_contact",
    fields=[
        {"source": "id", "target": "external_id"},
        {"source": "first_name", "target": "firstName"},
        {"source": "last_name", "target": "lastName"},
        {"source": "email", "target": "emailAddress"},
        {"source": "phone", "target": "phoneNumber", "transform": "format_phone"},
        # Custom field creation with default value
        {"target": "source", "value": "CRM Import"}
    ]
)

# Register a custom transformer for phone formatting
def format_phone(phone):
    if not phone:
        return ""
    # Remove non-digits
    digits = ''.join(c for c in phone if c.isdigit())
    # Format as (XXX) XXX-XXXX for US numbers
    if len(digits) == 10:
        return f"({digits[0:3]}) {digits[3:6]}-{digits[6:10]}"
    return phone

linker.mapper.register_transformer("format_phone", format_phone)

# Execute the sync
result = linker.sync()
print(f"Synced {result.count} customers to marketing platform")

2. Scheduled Data Collection

This example collects weather data hourly and saves to a CSV file:

from apilinker import ApiLinker
import csv
import datetime
import time
import os

# Create a function to handle the collected data
def save_weather_data(data, city):
    timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    # Create CSV if it doesn't exist
    file_exists = os.path.isfile(f"{city}_weather.csv")
    with open(f"{city}_weather.csv", mode='a', newline='') as file:
        writer = csv.writer(file)
        
        # Write header if file is new
        if not file_exists:
            writer.writerow(["timestamp", "temperature", "humidity", "conditions"])
        
        # Write data
        writer.writerow([
            timestamp,
            data['main']['temp'] - 273.15,  # Convert K to C
            data['main']['humidity'],
            data['weather'][0]['description']
        ])
    print(f"Weather data saved for {city} at {timestamp}")

# Initialize ApiLinker
linker = ApiLinker()

# Configure weather API
linker.add_source(
    type="rest",
    base_url="https://api.openweathermap.org/data/2.5",
    endpoints={
        "get_london_weather": {
            "path": "/weather",
            "method": "GET",
            "params": {
                "q": "London,uk",
                "appid": "YOUR_API_KEY"  # Replace with your API key
            }
        },
        "get_nyc_weather": {
            "path": "/weather",
            "method": "GET",
            "params": {
                "q": "New York,us",
                "appid": "YOUR_API_KEY"  # Replace with your API key
            }
        }
    }
)

# Create a custom handler for the weather data
def collect_weather():
    london_data = linker.fetch("get_london_weather")
    nyc_data = linker.fetch("get_nyc_weather")
    
    save_weather_data(london_data, "London")
    save_weather_data(nyc_data, "NYC")

# Run once to test
collect_weather()

# Then schedule to run hourly
linker.add_schedule(interval_minutes=60, callback=collect_weather)
linker.start_scheduled_sync()

# Keep the script running
try:
    print("Weather data collection started. Press Ctrl+C to stop.")
    while True:
        time.sleep(60)
except KeyboardInterrupt:
    print("Weather data collection stopped.")

🔌 Extending ApiLinker

Creating Custom Plugins

ApiLinker can be extended through plugins. Here's how to create a custom transformer plugin:

from apilinker.core.plugins import TransformerPlugin

class SentimentAnalysisTransformer(TransformerPlugin):
    """A transformer plugin that analyzes text sentiment."""
    
    plugin_name = "sentiment_analysis"  # This name is used to reference the plugin
    version = "1.0.0"                   # Optional version information
    author = "Your Name"                # Optional author information
    
    def transform(self, value, **kwargs):
        # Simple sentiment analysis (example)
        if not value or not isinstance(value, str):
            return {"sentiment": "neutral", "score": 0.0}
        
        # Add your sentiment analysis logic here
        positive_words = ["good", "great", "excellent"]
        negative_words = ["bad", "poor", "terrible"]
        
        # Count positive and negative words
        text = value.lower()
        positive_count = sum(1 for word in positive_words if word in text)
        negative_count = sum(1 for word in negative_words if word in text)
        
        # Calculate sentiment score
        total = positive_count + negative_count
        score = 0.0 if total == 0 else (positive_count - negative_count) / total
        
        return {
            "sentiment": "positive" if score > 0 else "negative" if score < 0 else "neutral",
            "score": score
        }

Using Your Custom Plugin

After creating your plugin, you need to register it before using:

from apilinker import ApiLinker

# Create your custom plugin instance
from my_plugins import SentimentAnalysisTransformer

# Initialize ApiLinker
linker = ApiLinker()

# Register the plugin
linker.plugin_manager.register_plugin(SentimentAnalysisTransformer)

# Configure APIs and mappings...
linker.add_mapping(
    source="get_reviews",
    target="save_analysis",
    fields=[
        {"source": "user_id", "target": "user_id"},
        # Use your custom plugin to transform the review text
        {"source": "review_text", "target": "sentiment_data", "transform": "sentiment_analysis"}
    ]
)

❓ Troubleshooting Guide

Installation Issues

• Package not found error

  ERROR: Could not find a version that satisfies the requirement apilinker
  

  • Make sure you're using Python 3.8 or newer
  • Check your internet connection
  • Try upgrading pip: pip install --upgrade pip

• Import errors

  ImportError: No module named 'apilinker'
  

  • Verify installation: pip list | grep apilinker
  • Check if you're using the correct Python environment
  • Try reinstalling: pip install --force-reinstall apilinker

Connection Issues

• API connection failures

  ConnectionError: Failed to establish connection to api.example.com
  

  • Check your internet connection
  • Verify the API base URL is correct
  • Make sure the API service is online
  • Check if your IP is allowed by the API provider

• Authentication errors

  AuthenticationError: Invalid credentials
  

  • Verify your API key or token is correct
  • Check if the token has expired
  • Ensure you're using the correct authentication method

Mapping Issues

• Field not found errors

  KeyError: 'Field not found in source data: user_profile'
  

  • Check the actual response data structure
  • Make sure you're referencing the correct field names
  • For nested fields, use dot notation (e.g., user.profile.name)

• Transformation errors

  ValueError: Invalid data for transformer 'iso_to_timestamp'
  

  • Check if the data matches the expected format
  • Make sure the transformer is properly registered
  • Add validation to your custom transformers

Common Code Examples

Handling API Rate Limits

from apilinker import ApiLinker
import time

linker = ApiLinker()

# Configure with retry settings
linker.add_source(
    type="rest",
    base_url="https://api.example.com",
    retry={
        "max_attempts": 5,
        "delay_seconds": 2,
        "backoff_factor": 2,  # Exponential backoff
        "status_codes": [429, 500, 502, 503, 504]  # Retry on these status codes
    },
    endpoints={
        "get_data": {"path": "/data", "method": "GET"}
    }
)

# Example of manual handling with wait periods
try:
    data = linker.fetch("get_data")
    print("Success!")
except Exception as e:
    if "rate limit" in str(e).lower():
        print("Rate limited, waiting and trying again...")
        time.sleep(60)  # Wait 1 minute
        data = linker.fetch("get_data")  # Try again
    else:
        raise e

📚 Documentation

Documentation is available in the /docs directory and will be hosted online soon.

Core Documentation

• Getting Started - A beginner-friendly introduction
• Installation Guide - Detailed installation instructions
• Configuration Guide - Configuration options and formats
• API Reference - Detailed API reference

Quick Resources

• Quick Reference - Essential commands and patterns
• FAQ - Frequently asked questions
• Troubleshooting Guide - Solutions to common problems

Guides and Examples

• Cookbook - Ready-to-use recipes for common tasks
• Examples - Example use cases and code
• Extending with Plugins - Creating and using plugins
• Security Considerations - Security best practices

Technical Documentation

• Architecture - System architecture and data flow diagrams
• Comparison - How ApiLinker compares to other integration tools

Step-by-Step Tutorials

• API-to-API Sync Tutorial - Learn to sync data between APIs
• Custom Transformers Tutorial - Create data transformation functions
• More tutorials - Browse all available tutorials

Comprehensive API Reference

For developers who want to extend ApiLinker or understand its internals, we provide comprehensive API reference documentation that can be generated using Sphinx:

# Install Sphinx and required packages
pip install sphinx sphinx-rtd-theme myst-parser

# Generate HTML documentation
cd docs/sphinx_setup
sphinx-build -b html . _build/html

The generated documentation will be available in docs/sphinx_setup/_build/html/index.html

Community Support

• GitHub Issues - Report bugs or request features
• Stack Overflow - Ask questions using the apilinker tag

🔒 Security Considerations

When working with APIs that require authentication, follow these security best practices:

• Never hardcode credentials in your code or configuration files. Always use environment variables or secure credential stores.

• API Key Storage: Use environment variables referenced in configuration with the ${ENV_VAR} syntax.

  auth:
    type: api_key
    header: X-API-Key
    key: ${MY_API_KEY}
  

• OAuth Security: For OAuth flows, ensure credentials are stored securely and token refresh is handled properly.

• Credential Validation: ApiLinker performs validation checks on authentication configurations to prevent common security issues.

• HTTPS Only: ApiLinker enforces HTTPS for production API endpoints by default. Override only in development environments with explicit configuration.

• Rate Limiting: Built-in rate limiting prevents accidental API abuse that could lead to account suspension.

• Audit Logging: Enable detailed logging for security-relevant events with:

  logging:
    level: INFO
    security_audit: true
  

🤝 Contributing

Contributions are welcome! Please see our Contributing Guide for details.

• Fork the repository
• Create your feature branch (git checkout -b feature/amazing-feature)
• Install development dependencies (pip install -e ".[dev]")
• Make your changes
• Run tests (pytest)
• Commit your changes (git commit -m 'Add amazing feature')
• Push to the branch (git push origin feature/amazing-feature)
• Open a Pull Request

📄 Citation

If you use ApiLinker in your research, please cite:

@software{apilinker2025,
  author = {Kartas, Kyriakos},
  title = {ApiLinker: A Universal Bridge for REST API Integrations},
  url = {https://github.com/kkartas/apilinker},
  version = {0.2.0},
  year = {2025},
  doi = {10.21105/joss.12345}
}

📃 License

This project is licensed under the MIT License - see the LICENSE file for details.