Oil prices in Python with source context

Install and authenticate

Run python -m pip install oilpriceapi. Create an API key in your dashboardand set OILPRICEAPI_KEY in your shell or secret manager. Do not put a key in source code, a notebook cell, a URL, logs, or screenshots. Check current plan accessfor the dataset you want.

Request Brent with recovery paths

Save this as oil_price.py and run it with the key in your environment. It calls the current /v1/prices/latestendpoint, checks the response shape, and prints the price with its source timestamp. Change BRENT_CRUDE_USD toWTI_USD for WTI. Both codes returned matching records from the live API on October 31, 2026.

import math
import os

from oilpriceapi import OilPriceAPI
from oilpriceapi.exceptions import (
    AuthenticationError, OilPriceAPIError, RateLimitError, TimeoutError
)

try:
    with OilPriceAPI(api_key=os.environ["OILPRICEAPI_KEY"], max_retries=1) as client:
        payload = client.request(
            "GET", "/v1/prices/latest",
            params={"by_code": "BRENT_CRUDE_USD"}, timeout=30,
        )
except AuthenticationError:
    raise SystemExit("API key missing, expired, or revoked. Replace it in your environment.")
except RateLimitError as error:
    raise SystemExit(f"Rate limited. Retry after {error.seconds_until_reset} seconds.")
except TimeoutError:
    raise SystemExit("Request timed out. Retry once, then check status.oilpriceapi.com.")
except OilPriceAPIError as error:
    if error.status_code in (402, 403):
        raise SystemExit("This account cannot access the requested dataset. Check your plan.")
    raise

record = payload.get("data")
if not isinstance(record, dict):
    raise SystemExit("No price record returned; do not substitute a zero.")

price = record.get("price")
if isinstance(price, bool) or not isinstance(price, (int, float)) or not math.isfinite(price):
    raise SystemExit("Price is missing or malformed; stop the calculation.")

source = record.get("source")
metadata = record.get("metadata")
if not source and isinstance(metadata, dict):
    source = metadata.get("source")
timestamp = next(
    (record[field] for field in ("as_of", "source_timestamp", "created_at")
     if isinstance(record.get(field), str) and record[field].strip()),
    None,
)
if not all((record.get("code"), record.get("currency"), record.get("unit"), source, timestamp)):
    raise SystemExit("Source context is incomplete; do not use this value.")

print({
    "code": record["code"], "price": price,
    "currency": record["currency"], "unit": record["unit"],
    "source": source, "source_timestamp": timestamp,
    "freshness": record.get("data_status") or record.get("freshness"),
})

The example makes one attempt and handles missing or revoked keys, rate limits, timeouts, entitlement errors, malformed price values, and missing source context. It never substitutes zero for missing data; inspect freshness before using a value. For other response codes and supported datasets, use the SDK documentation.

Use it in a daily workflow

Run the script once per day with your scheduler and store the returned value alongside its code, unit, source timestamp, and freshness state. A new request may return the same observation when its source has not published another one. Treat the source timestamp as the observation date, not the time your job ran. Do not loop every few seconds to force a new observation.

Prefer a different tool? See the Excel,Google Sheets, and Node.js tutorials.

For endpoints and response fields, see the API documentation.