Python SDK

Official Python SDK for OilPriceAPI - Simple, intuitive, and Pythonic.

(opens new window) (opens new window) (opens new window) (opens new window)

📦 Installation

# Using pip
pip install oilpriceapi

# Using pip with extras for async support
pip install oilpriceapi[async]

# Using poetry
poetry add oilpriceapi

# Using conda
conda install -c conda-forge oilpriceapi

Requirements

• Python 3.7+
• requests library (installed automatically)
• aiohttp for async support (optional)

🚀 Quick Start

Basic Usage

from oilpriceapi import Client

# Initialize client
client = Client(api_key="YOUR_API_KEY")

# Get latest prices
prices = client.prices.get_latest()
print(f"WTI: ${prices['data']['WTI']['price']}")
print(f"Brent: ${prices['data']['BRENT']['price']}")

Using Environment Variables

import os
from oilpriceapi import Client

# Set environment variable
# export OILPRICE_API_KEY="your_api_key"

# Client will auto-detect the API key
client = Client()

# Get specific commodities
oil_prices = client.prices.get_latest(commodities=["WTI", "BRENT"])

Async Support

import asyncio
from oilpriceapi.async_client import AsyncClient

async def get_prices():
    async with AsyncClient(api_key="YOUR_API_KEY") as client:
        prices = await client.prices.get_latest()
        return prices

# Run async function
prices = asyncio.run(get_prices())

🔧 Configuration

Client Options

from oilpriceapi import Client

client = Client(
    api_key="YOUR_API_KEY",              # Required unless in env
    base_url="https://api.oilpriceapi.com",  # Optional
    timeout=30,                          # Request timeout in seconds
    max_retries=3,                       # Automatic retry on failure
    retry_delay=1.0,                     # Delay between retries
    verify_ssl=True,                     # SSL certificate verification
    proxies={                            # Optional proxy configuration
        "http": "http://proxy.example.com:8080",
        "https": "https://proxy.example.com:8080"
    }
)

Environment Configuration

# .env file
OILPRICE_API_KEY=opa_live_a1b2c3d4e5f6
OILPRICE_API_URL=https://api.oilpriceapi.com
OILPRICE_API_TIMEOUT=30

from oilpriceapi import Client
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Client auto-configures from env
client = Client()

📚 API Methods

Prices

Get Latest Prices

# Get specific commodities (required)
oil_prices = client.prices.get_latest(
    by_code=["WTI_USD", "BRENT_CRUDE_USD", "DUBAI_CRUDE_USD"]
)

# Get prices in different currency
eur_prices = client.prices.get_latest(
    by_code=["WTI_USD", "BRENT_CRUDE_USD"],
    currency="EUR"
)

# Get specific fields only
minimal = client.prices.get_latest(
    by_code=["WTI_USD"],
    fields=["value", "timestamp"]
)

# Pretty print the response
from pprint import pprint
pprint(oil_prices)

Get Historical Prices

# Past 24 hours with hourly data
daily = client.prices.get_past_day(
    by_code=["WTI_USD", "BRENT_CRUDE_USD"]
)

# Past week with daily data
weekly = client.prices.get_past_week(
    by_code=["WTI_USD"],
    interval="1d"
)

# Past month
monthly = client.prices.get_past_month(
    by_code=["NATURAL_GAS_USD"]
)

# Past year with weekly data
yearly = client.prices.get_past_year(
    by_code=["WTI_USD", "BRENT_CRUDE_USD"],
    interval="1w"
)

# Custom date range (requires premium plan)
historical = client.prices.get_historical(
    start_date="2024-01-01",
    end_date="2024-12-31",
    by_code=["WTI_USD"],
    interval="1d"
)

Commodities

# Get all available commodities
commodities = client.commodities.get_all()
for commodity in commodities['data']:
    print(f"{commodity['code']}: {commodity['name']} ({commodity['unit']})")

# Get specific commodity details
wti_info = client.commodities.get("WTI_USD")

# Get commodities by category
categories = client.commodities.get_categories()
crude_oils = client.commodities.get_by_category("crude_oil")

Data Analysis

import pandas as pd
import matplotlib.pyplot as plt

# Convert to pandas DataFrame
prices_data = client.prices.get_past_week(by_code=["WTI_USD", "BRENT_CRUDE_USD"])

# Create DataFrame
df_wti = pd.DataFrame(prices_data['data']['prices']['WTI_USD']['prices'])
df_wti['timestamp'] = pd.to_datetime(df_wti['timestamp'])
df_wti.set_index('timestamp', inplace=True)

# Plot the data
plt.figure(figsize=(10, 6))
plt.plot(df_wti.index, df_wti['value'])
plt.title('WTI Oil Price - Past Week')
plt.xlabel('Date')
plt.ylabel('Price (USD)')
plt.grid(True)
plt.show()

# Calculate statistics
print(f"Average: ${df_wti['value'].mean():.2f}")
print(f"Min: ${df_wti['value'].min():.2f}")
print(f"Max: ${df_wti['value'].max():.2f}")
print(f"Std Dev: ${df_wti['value'].std():.2f}")

WebSocket Streaming (Reservoir Mastery)

from oilpriceapi import StreamClient

def on_price_update(data):
    """Handle real-time price updates"""
    print(f"{data['commodity']}: ${data['price']} "
          f"({'+' if data['change'] > 0 else ''}{data['change']})")

def on_error(error):
    print(f"Stream error: {error}")

# Create streaming client
stream = StreamClient(api_key="YOUR_API_KEY")

# Connect to stream
stream.connect(
    commodities=["WTI_USD", "BRENT_CRUDE_USD", "NATURAL_GAS_USD"],
    on_price=on_price_update,
    on_error=on_error
)

# Keep running
try:
    stream.run_forever()
except KeyboardInterrupt:
    stream.disconnect()

Price Alerts

# Create a price alert
alert = client.alerts.create(
    commodity="WTI_USD",
    condition="above",
    threshold=80.00,
    notification_email=True,
    notification_webhook="https://your-app.com/webhook"
)

# List all active alerts
my_alerts = client.alerts.list()
for alert in my_alerts['data']:
    print(f"Alert {alert['id']}: {alert['commodity']} "
          f"{alert['condition']} ${alert['threshold']}")

# Update an alert
client.alerts.update(
    alert_id=alert['id'],
    threshold=85.00
)

# Delete an alert
client.alerts.delete(alert['id'])

# Test alert notification
client.alerts.test(alert['id'])

Account Management

# Get account information
account = client.account.get()
print(f"Plan: {account['plan']}")
print(f"Requests used: {account['requests_used']}/{account['request_limit']}")
print(f"Reset date: {account['reset_date']}")

# Get usage statistics
usage = client.account.get_usage(period="current_month")
print(f"Total requests: {usage['total_requests']}")
print(f"By endpoint:")
for endpoint, count in usage['by_endpoint'].items():
    print(f"  {endpoint}: {count}")

# Get API keys
api_keys = client.account.get_api_keys()
for key in api_keys['data']:
    print(f"Key: {key['key'][:10]}... Created: {key['created_at']}")

🛡️ Error Handling

Basic Error Handling

from oilpriceapi import Client, OilPriceAPIError

client = Client(api_key="YOUR_API_KEY")

try:
    prices = client.prices.get_latest()
except OilPriceAPIError as e:
    print(f"API Error: {e}")
    print(f"Status Code: {e.status_code}")
    print(f"Error Code: {e.error_code}")
    print(f"Details: {e.details}")

Specific Error Types

from oilpriceapi.exceptions import (
    RateLimitError,
    AuthenticationError,
    ValidationError,
    NetworkError,
    ServerError
)

try:
    prices = client.prices.get_latest()
except RateLimitError as e:
    print(f"Rate limit exceeded. Reset at: {e.reset_at}")
    print(f"Retry after: {e.retry_after} seconds")
except AuthenticationError:
    print("Invalid API key")
except ValidationError as e:
    print(f"Invalid parameters: {e.details}")
except NetworkError:
    print("Network connection issue")
except ServerError:
    print("Server error, please try again")

Retry Logic

import time
from oilpriceapi import Client

def get_prices_with_retry(client, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.prices.get_latest()
        except RateLimitError as e:
            if attempt < max_retries - 1:
                time.sleep(e.retry_after)
            else:
                raise
        except (NetworkError, ServerError) as e:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
            else:
                raise

💡 Advanced Usage

Caching Responses

import json
import time
from pathlib import Path

class CachedClient:
    def __init__(self, api_key, cache_dir="cache", cache_time=300):
        self.client = Client(api_key=api_key)
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
        self.cache_time = cache_time
    
    def get_latest_prices(self, commodities=None):
        cache_key = f"prices_{'_'.join(commodities) if commodities else 'all'}.json"
        cache_file = self.cache_dir / cache_key
        
        # Check cache
        if cache_file.exists():
            with open(cache_file) as f:
                cached = json.load(f)
                if time.time() - cached['timestamp'] < self.cache_time:
                    return cached['data']
        
        # Fetch fresh data
        data = self.client.prices.get_latest(commodities=commodities)
        
        # Update cache
        with open(cache_file, 'w') as f:
            json.dump({
                'timestamp': time.time(),
                'data': data
            }, f)
        
        return data

Pandas Integration

import pandas as pd
from datetime import datetime, timedelta

class OilPriceDataFrame:
    def __init__(self, client):
        self.client = client
    
    def get_price_df(self, commodities, days=7):
        """Get price data as pandas DataFrame"""
        if days <= 1:
            data = self.client.prices.get_past_day(commodities=commodities)
        elif days <= 7:
            data = self.client.prices.get_past_week(commodities=commodities)
        elif days <= 30:
            data = self.client.prices.get_past_month(commodities=commodities)
        else:
            data = self.client.prices.get_past_year(commodities=commodities)
        
        # Convert to DataFrame
        dfs = []
        for commodity, prices in data['data'].items():
            df = pd.DataFrame(prices)
            df['commodity'] = commodity
            df['timestamp'] = pd.to_datetime(df['timestamp'])
            dfs.append(df)
        
        return pd.concat(dfs, ignore_index=True)
    
    def get_price_comparison(self, commodities=['WTI', 'BRENT']):
        """Compare multiple commodity prices"""
        df = self.get_price_df(commodities, days=30)
        
        # Pivot for comparison
        pivot = df.pivot(index='timestamp', columns='commodity', values='price')
        
        # Calculate correlation
        correlation = pivot.corr()
        
        return pivot, correlation

# Usage
oil_df = OilPriceDataFrame(client)
prices, corr = oil_df.get_price_comparison(['WTI', 'BRENT'])
print(f"WTI-Brent Correlation: {corr.loc['WTI', 'BRENT']:.3f}")

Async Batch Operations

import asyncio
from oilpriceapi.async_client import AsyncClient

async def fetch_multiple_periods(api_key):
    async with AsyncClient(api_key=api_key) as client:
        # Fetch multiple time periods concurrently
        tasks = [
            client.prices.get_latest(),
            client.prices.get_past_day(),
            client.prices.get_past_week(),
            client.commodities.get_all()
        ]
        
        results = await asyncio.gather(*tasks)
        
        return {
            'latest': results[0],
            'daily': results[1],
            'weekly': results[2],
            'commodities': results[3]
        }

# Run async function
data = asyncio.run(fetch_multiple_periods("YOUR_API_KEY"))

Command Line Interface

# oilprice_cli.py
import click
from oilpriceapi import Client
from tabulate import tabulate

@click.command()
@click.option('--commodities', '-c', multiple=True, help='Commodity codes')
@click.option('--currency', default='USD', help='Currency for prices')
@click.option('--format', type=click.Choice(['table', 'json', 'csv']), default='table')
def get_prices(commodities, currency, format):
    """Get latest oil prices"""
    client = Client()
    
    prices = client.prices.get_latest(
        commodities=list(commodities) if commodities else None,
        currency=currency
    )
    
    if format == 'table':
        data = []
        for code, info in prices['data'].items():
            data.append([
                code,
                f"{info['price']:.2f}",
                currency,
                f"{info['change_percent_24h']:+.2f}%"
            ])
        
        print(tabulate(data, headers=['Commodity', 'Price', 'Currency', 'Change']))
    
    elif format == 'json':
        import json
        print(json.dumps(prices, indent=2))
    
    elif format == 'csv':
        print("commodity,price,currency,change_percent")
        for code, info in prices['data'].items():
            print(f"{code},{info['price']},{currency},{info['change_percent_24h']}")

if __name__ == '__main__':
    get_prices()

🧪 Testing

Unit Tests

# test_oilprice.py
import unittest
from unittest.mock import Mock, patch
from oilpriceapi import Client

class TestOilPriceClient(unittest.TestCase):
    def setUp(self):
        self.client = Client(api_key="test_key")
    
    @patch('oilpriceapi.client.requests.get')
    def test_get_latest_prices(self, mock_get):
        # Mock response
        mock_response = Mock()
        mock_response.status_code = 200
        mock_response.json.return_value = {
            'status': 'success',
            'data': {
                'WTI': {'price': 75.50}
            }
        }
        mock_get.return_value = mock_response
        
        # Test
        prices = self.client.prices.get_latest()
        
        # Assertions
        self.assertEqual(prices['data']['WTI']['price'], 75.50)
        mock_get.assert_called_once()

if __name__ == '__main__':
    unittest.main()

Integration Tests

# test_integration.py
import pytest
from oilpriceapi import Client

@pytest.fixture
def client():
    return Client()  # Uses API key from environment

def test_get_real_prices(client):
    """Test with real API"""
    prices = client.prices.get_latest(commodities=["WTI"])
    
    assert prices['status'] == 'success'
    assert 'WTI' in prices['data']
    assert prices['data']['WTI']['price'] > 0

def test_rate_limit_headers(client):
    """Check rate limit headers"""
    response = client.prices.get_latest()
    
    # Check response includes rate limit info
    assert 'X-RateLimit-Limit' in response.get('headers', {})

📊 Examples

Flask Web App

from flask import Flask, jsonify
from oilpriceapi import Client
import os

app = Flask(__name__)
client = Client(api_key=os.getenv('OILPRICE_API_KEY'))

@app.route('/api/prices')
def get_prices():
    try:
        prices = client.prices.get_latest()
        return jsonify(prices)
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/prices/<commodity>')
def get_commodity_price(commodity):
    try:
        prices = client.prices.get_latest(commodities=[commodity.upper()])
        return jsonify(prices['data'][commodity.upper()])
    except Exception as e:
        return jsonify({'error': str(e)}), 404

if __name__ == '__main__':
    app.run(debug=True)

Jupyter Notebook

# In Jupyter Notebook
%matplotlib inline

import matplotlib.pyplot as plt
import seaborn as sns
from oilpriceapi import Client
import pandas as pd

# Initialize client
client = Client()

# Get weekly data
weekly_data = client.prices.get_past_week(
    commodities=['WTI', 'BRENT', 'NATURAL_GAS']
)

# Create DataFrame
dfs = []
for commodity, prices in weekly_data['data'].items():
    df = pd.DataFrame(prices)
    df['commodity'] = commodity
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    dfs.append(df)

df_all = pd.concat(dfs)

# Plot
plt.figure(figsize=(12, 6))
sns.lineplot(data=df_all, x='timestamp', y='price', hue='commodity')
plt.title('Oil & Gas Prices - Past Week')
plt.xlabel('Date')
plt.ylabel('Price (USD)')
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

📚 Additional Examples

Find more examples in our GitHub repository (opens new window):

• Django Integration (opens new window)
• FastAPI Service (opens new window)
• Data Analysis (opens new window)
• Trading Bot (opens new window)
• Telegram Bot (opens new window)
• AWS Lambda (opens new window)

🔗 Resources

• API Documentation (opens new window)
• PyPI Package (opens new window)
• GitHub Repository (opens new window)
• Support

---

Need help? Join our Discord (opens new window) or open an issue (opens new window)