Prerequisites

• Completion of Lesson 1 (environment & sandbox setup)
• Python 3.10+, requests installed, a .env file with BETFAIR_API_URL and BETFAIR_API_KEY
• Basics of HTTP status codes and JSON

Overview

We’ll build an auth wrapper that reads credentials from environment variables, authenticates to the Betfair sandbox (or delayed/read-only endpoint), refreshes tokens when necessary, and provides a robust request() helper with retries and backoff.

1) Recommended .env values

Keep these in your .env file (never commit). The download button above provides a template.

Copy# .env
BETFAIR_API_URL=https://sandbox-api.example.com
BETFAIR_API_KEY=your_api_key_here
BETFAIR_USERNAME=test_user
BETFAIR_PASSWORD=test_pass
# Optional: token storage path
TOKEN_STORE=./.betfair_token.json

2) Auth wrapper (auth.py)

Save this as auth.py. It demonstrates: loading env, a minimal login flow (if your provider requires username/password), token caching, automatic refresh on 401, and a resilient request helper. Adapt headers and endpoints to your Betfair API variant.

Copyimport os
import json
import time
import requests
from dotenv import load_dotenv
from pathlib import Path

load_dotenv()

API_URL = os.getenv('BETFAIR_API_URL')  # e.g. sandbox market/data endpoint
API_KEY = os.getenv('BETFAIR_API_KEY')
TOKEN_STORE = Path(os.getenv('TOKEN_STORE', '.betfair_token.json'))

HEADERS_BASE = {
    'X-Application': API_KEY,
    'Content-Type': 'application/json'
}

# Simple token cache
def load_token():
    if TOKEN_STORE.exists():
        try:
            return json.loads(TOKEN_STORE.read_text())
        except Exception:
            return None
    return None

def save_token(obj):
    TOKEN_STORE.write_text(json.dumps(obj))

def authenticate(username=None, password=None):
    """
    Example: Many Betfair APIs use an API key plus a session token.
    Replace this with your provider's auth flow. This stub returns a fake token.
    """
    # Replace with real auth call if required by provider
    now = int(time.time())
    token_obj = {
        "sessionToken": "sandbox_dummy_token_" + str(now),
        "expires_at": now + 3600
    }
    save_token(token_obj)
    return token_obj

def get_valid_token():
    t = load_token()
    if not t or t.get('expires_at', 0) <= int(time.time()):
        # Re-authenticate — you may need username/password or a client-credentials flow
        t = authenticate(os.getenv('BETFAIR_USERNAME'), os.getenv('BETFAIR_PASSWORD'))
    return t

# resilient request helper with retries/backoff and auto-refresh on 401
def request(method, path, params=None, json_body=None, max_retries=3, backoff_factor=0.6, timeout=10):
    url = API_URL.rstrip('/') + '/' + path.lstrip('/')
    for attempt in range(1, max_retries + 1):
        token = get_valid_token()
        headers = dict(HEADERS_BASE)
        if token and token.get('sessionToken'):
            headers['X-Authentication'] = token['sessionToken']
        try:
            resp = requests.request(method, url, headers=headers, params=params, json=json_body, timeout=timeout)
            if resp.status_code == 401:
                # token may be expired/invalid — refresh and retry immediately (once)
                authenticate(os.getenv('BETFAIR_USERNAME'), os.getenv('BETFAIR_PASSWORD'))
                token = get_valid_token()
                headers['X-Authentication'] = token.get('sessionToken')
                resp = requests.request(method, url, headers=headers, params=params, json=json_body, timeout=timeout)
            resp.raise_for_status()
            return resp.json()
        except requests.HTTPError as e:
            status = getattr(e.response, 'status_code', None)
            if status and 500 <= status < 600 and attempt < max_retries:
                # server error - retry
                time.sleep(backoff_factor * (2 ** (attempt - 1)))
                continue
            # non-retryable or max attempts exhausted
            raise
        except requests.RequestException:
            if attempt < max_retries:
                time.sleep(backoff_factor * (2 ** (attempt - 1)))
                continue
            raise

Notes: Replace the authenticate() body with Betfair's actual login/token exchange (or delayed API key usage). Always avoid storing plaintext keys in repos — use the .env file for local development and a secrets manager in production.

3) Example usage

Use the request helper to fetch markets or account info. Example:

Copyfrom auth import request

# simple market search (adjust path/params to provider)
markets = request('GET', '/markets', params={'filter':'Match Odds'})
print('Found', len(markets) if isinstance(markets, list) else 'response', markets)

Tip: Inspect returned JSON in sandbox — real responses vary between Betfair data API versions. If you see 401/403, check your API key and session token flow.

4) Token storage & security

• Store tokens locally only if necessary (development). For production, use a secure secret store or encrypted DB.
• Set tight file permissions on token stores (chmod 600) and avoid committing them.
• Rotate keys and tokens periodically; detect abnormal failures (many 401s) and invalidate tokens.

5) Idempotency & retries for order endpoints

When you later place orders, use idempotency tokens so retries don't duplicate orders. Example approach: generate a UUID per trading signal and include it in your order metadata; on replays check server-side before creating a duplicate order.

What you'll build next

Lesson 3 will use this auth helper to search markets and pick runner IDs programmatically. Lesson 4 will demonstrate order placement in the sandbox using idempotency and retry-safe logic.

Back to course hub