Python Examples

Complete Python examples for integrating OpenFXRates using the Requests library and async patterns.

Installation

Using pip

pip install requests python-dotenv

For async examples:

pip install aiohttp python-dotenv

Example 1: Get Latest Rates

import requests
import os
from dotenv import load_dotenv

load_dotenv()

api_key = os.getenv('OPENFXRATES_API_KEY')
api_url = 'https://api.openfxrates.com'

def get_latest_rates(base, targets=None):
    """Fetch the latest exchange rates."""
    
    params = {
        'base': base,
    }
    if targets:
        params['targets'] = targets

    headers = {
        'X-API-Key': api_key,
        'Content-Type': 'application/json'
    }

    try:
        response = requests.get(
            f'{api_url}/latest_rates',
            params=params,
            headers=headers,
            timeout=30
        )
        response.raise_for_status()

        data = response.json()

        print(f"Base: {data['base']}")
        print(f"Date: {data['date']}")
        print("Rates:")

        for currency, rate in data['rates'].items():
            print(f"  {currency}: {rate:.4f}")

        return data

    except requests.exceptions.RequestException as e:
        print(f"Error fetching rates: {e}")
        return None

# Usage
get_latest_rates('USD', 'EUR,GBP,JPY')

Example 2: Convert Currency

def convert_currency(from_currency, to_currencies, amount=1):
    """Convert an amount from one currency to target currencies."""
    
    params = {
        'from': from_currency,
        'to': to_currencies,
        'amount': amount
    }

    headers = {'X-API-Key': api_key}

    try:
        response = requests.get(
            f'{api_url}/convert',
            params=params,
            headers=headers
        )
        response.raise_for_status()

        data = response.json()

        print(f"Converting: {amount} {data['from']}")
        print("Results:")

        for currency, converted in data['conversions'].items():
            print(f"  {currency}: {converted:.2f}")

        return data

    except requests.exceptions.RequestException as e:
        print(f"Conversion error: {e}")
        return None

# Usage
convert_currency('USD', 'EUR,GBP,JPY', 100)

Example 3: Get Historical Rates

def get_historical_rates(base, date, targets=None):
    """Fetch historical exchange rates for a specific date."""
    
    params = {
        'base': base,
        'date': date,
    }
    if targets:
        params['targets'] = targets

    headers = {'X-API-Key': api_key}

    try:
        response = requests.get(
            f'{api_url}/historical_rates',
            params=params,
            headers=headers
        )
        response.raise_for_status()

        data = response.json()

        print(f"Historical rates for {data['date']}")
        print(f"Base: {data['base']}")
        print("Rates:")

        for currency, rate in data['rates'].items():
            print(f"  {currency}: {rate:.4f}")

        return data

    except requests.exceptions.HTTPError as e:
        error_data = e.response.json()
        print(f"Error: {error_data.get('message', 'Unknown error')}")
        return None

# Usage - Get rates from January 15, 2024
get_historical_rates('USD', '2024-01-15', 'EUR,GBP')

Example 4: List All Currencies

def list_currencies():
    """Fetch all available currencies."""
    
    headers = {'X-API-Key': api_key}

    try:
        response = requests.get(
            f'{api_url}/currencies',
            headers=headers
        )
        response.raise_for_status()

        data = response.json()

        print(f"Available currencies: {len(data['currencies'])}\n")

        for currency in data['currencies']:
            symbol = f" ({currency['symbol']})" if 'symbol' in currency else ""
            print(f"{currency['code']} - {currency['name']}{symbol}")

        return data['currencies']

    except requests.exceptions.RequestException as e:
        print(f"Error listing currencies: {e}")
        return None

# Usage
list_currencies()

Example 5: API Client Class

import requests
from typing import Optional, Dict, Any
import os
from dotenv import load_dotenv

load_dotenv()

class OpenFXRatesClient:
    """Client for interacting with OpenFXRates API."""

    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv('OPENFXRATES_API_KEY')
        self.base_url = 'https://api.openfxrates.com'
        self.session = requests.Session()
        self.session.headers.update({
            'X-API-Key': self.api_key,
            'Content-Type': 'application/json'
        })

    def get_latest_rates(self, base: str, targets: Optional[str] = None) -> Optional[Dict]:
        """Get the latest exchange rates."""
        return self._request('/latest_rates', {
            'base': base,
            'targets': targets
        })

    def get_historical_rates(self, base: str, date: str, targets: Optional[str] = None) -> Optional[Dict]:
        """Get historical exchange rates for a specific date."""
        return self._request('/historical_rates', {
            'base': base,
            'date': date,
            'targets': targets
        })

    def convert_currency(self, from_currency: str, to_currencies: str, amount: float = 1) -> Optional[Dict]:
        """Convert currency amount."""
        return self._request('/convert', {
            'from': from_currency,
            'to': to_currencies,
            'amount': amount
        })

    def list_currencies(self) -> Optional[Dict]:
        """List all available currencies."""
        return self._request('/currencies')

    def _request(self, endpoint: str, params: Dict = None) -> Optional[Dict]:
        """Make a request to the API."""
        try:
            response = self.session.get(
                f'{self.base_url}{endpoint}',
                params=params,
                timeout=30
            )
            response.raise_for_status()
            return response.json()

        except requests.exceptions.RequestException as e:
            print(f"Request error: {e}")
            return None

# Usage
client = OpenFXRatesClient()

rates = client.get_latest_rates('USD', 'EUR,GBP,JPY')
if rates:
    print(f"Base: {rates['base']}")
    print(f"Rates: {rates['rates']}")

Example 6: Error Handling and Logging

import requests
import logging
from typing import Optional, Dict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class RobustOpenFXRatesClient:
    """Client with comprehensive error handling."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = 'https://api.openfxrates.com'

    def get_latest_rates(self, base: str, targets: str) -> Optional[Dict]:
        """Get latest rates with error handling."""
        try:
            response = requests.get(
                f'{self.base_url}/latest_rates',
                params={'base': base, 'targets': targets},
                headers={'X-API-Key': self.api_key},
                timeout=30
            )

            # Handle HTTP errors
            if response.status_code == 401:
                logger.error('Unauthorized: Invalid API key')
                return None

            elif response.status_code == 400:
                error = response.json()
                logger.error(f"Bad request: {error.get('message')}")
                return None

            elif response.status_code == 429:
                retry_after = response.headers.get('retry-after', 60)
                logger.warning(f"Rate limited. Retry after {retry_after}s")
                return None

            elif response.status_code == 200:
                logger.info(f"Successfully fetched rates for {base}")
                return response.json()

            else:
                logger.error(f"Unexpected status code: {response.status_code}")
                return None

        except requests.exceptions.Timeout:
            logger.error("Request timeout")
            return None

        except requests.exceptions.ConnectionError:
            logger.error("Connection error")
            return None

        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            return None

# Usage
client = RobustOpenFXRatesClient(os.getenv('OPENFXRATES_API_KEY'))
rates = client.get_latest_rates('USD', 'EUR,GBP')

Example 7: Caching with TTL

from datetime import datetime, timedelta
import json

class CachedOpenFXRatesClient:
    """Client with built-in caching."""

    def __init__(self, api_key: str, cache_ttl: int = 3600):
        self.api_key = api_key
        self.base_url = 'https://api.openfxrates.com'
        self.cache = {}
        self.cache_ttl = cache_ttl  # seconds

    def _cache_key(self, endpoint: str, params: Dict) -> str:
        """Generate a cache key."""
        return f"{endpoint}:{json.dumps(params, sort_keys=True)}"

    def _get_cached(self, key: str) -> Optional[Dict]:
        """Get value from cache if not expired."""
        if key in self.cache:
            data, timestamp = self.cache[key]
            if datetime.now() - timestamp < timedelta(seconds=self.cache_ttl):
                print(f"Cache hit: {key}")
                return data
            else:
                # Expired
                del self.cache[key]

        return None

    def get_latest_rates(self, base: str, targets: str = None) -> Optional[Dict]:
        """Get latest rates with caching."""
        
        params = {'base': base}
        if targets:
            params['targets'] = targets

        cache_key = self._cache_key('/latest_rates', params)

        # Check cache
        cached_data = self._get_cached(cache_key)
        if cached_data:
            return cached_data

        # Fetch from API
        try:
            response = requests.get(
                f'{self.base_url}/latest_rates',
                params=params,
                headers={'X-API-Key': self.api_key},
                timeout=30
            )
            response.raise_for_status()

            data = response.json()

            # Cache the result
            self.cache[cache_key] = (data, datetime.now())

            return data

        except requests.exceptions.RequestException as e:
            print(f"Error: {e}")
            return None

    def clear_cache(self):
        """Clear the cache."""
        self.cache.clear()

# Usage
client = CachedOpenFXRatesClient(os.getenv('OPENFXRATES_API_KEY'))

# First call - fetches from API
rates1 = client.get_latest_rates('USD', 'EUR,GBP')

# Second call - returns from cache
rates2 = client.get_latest_rates('USD', 'EUR,GBP')

# Clear cache when needed
client.clear_cache()

Example 8: Async/Await with aiohttp

import aiohttp
import asyncio
import os
from dotenv import load_dotenv

load_dotenv()

class AsyncOpenFXRatesClient:
    """Async client for OpenFXRates API."""

    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv('OPENFXRATES_API_KEY')
        self.base_url = 'https://api.openfxrates.com'

    async def get_latest_rates(self, base: str, targets: str = None):
        """Asynchronously fetch latest rates."""
        
        params = {'base': base}
        if targets:
            params['targets'] = targets

        headers = {'X-API-Key': self.api_key}

        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(
                    f'{self.base_url}/latest_rates',
                    params=params,
                    headers=headers,
                    timeout=aiohttp.ClientTimeout(total=30)
                ) as response:
                    if response.status == 200:
                        return await response.json()
                    else:
                        print(f"Error: HTTP {response.status}")
                        return None

            except aiohttp.ClientError as e:
                print(f"Request error: {e}")
                return None

    async def get_multiple_rates(self, bases: list, targets: str = None):
        """Fetch rates for multiple base currencies concurrently."""
        
        tasks = [
            self.get_latest_rates(base, targets)
            for base in bases
        ]

        return await asyncio.gather(*tasks)

# Usage
async def main():
    client = AsyncOpenFXRatesClient()

    # Single request
    rates = await client.get_latest_rates('USD', 'EUR,GBP')
    print("USD rates:", rates)

    # Multiple concurrent requests
    results = await client.get_multiple_rates(
        ['USD', 'EUR', 'GBP'],
        'JPY'
    )
    print("Multiple results:", results)

# Run
asyncio.run(main())

Example 9: Pandas DataFrame Integration

import pandas as pd
import os
from dotenv import load_dotenv

load_dotenv()

class DataFrameOpenFXRatesClient:
    """Client that returns Pandas DataFrames."""

    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv('OPENFXRATES_API_KEY')
        self.base_url = 'https://api.openfxrates.com'

    def get_rates_dataframe(self, base: str, targets: str = None) -> pd.DataFrame:
        """Get rates as a Pandas DataFrame."""
        
        response = requests.get(
            f'{self.base_url}/latest_rates',
            params={'base': base, 'targets': targets},
            headers={'X-API-Key': self.api_key}
        )
        response.raise_for_status()

        data = response.json()

        # Create DataFrame
        df = pd.DataFrame({
            'currency': list(data['rates'].keys()),
            'rate': list(data['rates'].values()),
            'base': data['base'],
            'date': data['date']
        })

        return df

    def get_historical_dataframe(self, base: str, start_date: str, end_date: str, targets: str = None) -> pd.DataFrame:
        """Get historical rates as a DataFrame."""
        
        # Generate date range
        date_range = pd.date_range(start=start_date, end=end_date)
        
        rates_data = []

        for date in date_range:
            date_str = date.strftime('%Y-%m-%d')
            
            response = requests.get(
                f'{self.base_url}/historical_rates',
                params={'base': base, 'date': date_str, 'targets': targets},
                headers={'X-API-Key': self.api_key}
            )

            if response.status_code == 200:
                data = response.json()
                for currency, rate in data['rates'].items():
                    rates_data.append({
                        'date': date_str,
                        'base': base,
                        'currency': currency,
                        'rate': rate
                    })

        return pd.DataFrame(rates_data)

# Usage
client = DataFrameOpenFXRatesClient()

# Get current rates as DataFrame
df_current = client.get_rates_dataframe('USD', 'EUR,GBP,JPY')
print(df_current)

# Get historical rates
df_historical = client.get_historical_dataframe(
    'USD',
    '2024-01-01',
    '2024-01-31',
    'EUR'
)
print(df_historical)

# Calculate statistics
print("Rate statistics:")
print(df_historical.groupby('currency')['rate'].describe())

Best Practices

1. Use Environment Variables

   from dotenv import load_dotenv
   api_key = os.getenv('OPENFXRATES_API_KEY')

2. Implement Error Handling

   try:
       response.raise_for_status()
   except requests.exceptions.HTTPError as e:
       logger.error(f"HTTP error: {e}")

3. Use Sessions for Multiple Requests

   session = requests.Session()
   session.headers.update({'X-API-Key': api_key})

4. Add Timeouts

   response = requests.get(url, timeout=30)

5. Cache Responses

   # Cache for 1 hour

Next Steps

• PHP Examples → - PHP examples
• JavaScript Examples → - JavaScript and Node.js
• API Reference → - Full API documentation