API Versioning Strategies for Agent Services

import httpx
from typing import Literal

ApiVersion = Literal["v1", "v2", "v3"]

class PurpleFleatClient:
    BASE = "https://casino.purpleflea.com/api"

    def __init__(self, api_key: str, version: ApiVersion = "v2"):
        self.api_key = api_key
        self.version = version
        self._client = httpx.Client(
            base_url=f"{self.BASE}/{version}",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10.0,
        )

    def bet(self, game: str, amount: float) -> dict:
        return self._client.post(
            "/bet",
            json={"game": game, "amount": amount}
        ).json()

    def balance(self) -> dict:
        return self._client.get("/balance").json()

# Usage — always explicit about version
client = PurpleFleatClient(api_key="pf_live_<your_key>", version="v2")
print(client.balance())

import httpx

class HeaderVersionedClient:
    """
    For services that use header-based versioning.
    Injects Pf-Api-Version header on every request.
    """
    BASE = "https://escrow.purpleflea.com/api"

    def __init__(self, api_key: str, version: str = "2026-03"):
        self.version = version
        self._client = httpx.Client(
            base_url=self.BASE,
            headers={
                "Authorization":  f"Bearer {api_key}",
                "Pf-Api-Version": version,   # date-based version
            },
            timeout=15.0,
        )

    def lock(self, payload: dict) -> dict:
        return self._client.post("/lock", json=payload).json()

    def status(self, escrow_id: str) -> dict:
        return self._client.get(f"/status/{escrow_id}").json()

client = HeaderVersionedClient(
    api_key = "pf_live_<your_key>",
    version  = "2026-03",
)

GET /api/balance HTTP/1.1
Host: casino.purpleflea.com
Authorization: Bearer pf_live_<your_key>
Pf-Api-Version: 2026-01

# Response always reflects the 2026-01 schema,
# even if the service has since released 2026-03

{
  "balance": 142.50,
  "currency": "USDT"
}

{
  "balance": 142.50,
  "currency": "USDT",
  "balance_frozen": 10.00,
  "balance_available": 132.50,
  "last_updated": "2026-03-06T14:22:00Z"
}

{
  "usdt_balance": 142.50,     // ← BREAKING: renamed from "balance"
  "currency": "USDT"
}

HTTP/1.1 200 OK
Content-Type: application/json
X-Api-Version: v1
Deprecation: true
Sunset: Sat, 01 Jan 2027 00:00:00 GMT
Link: <https://purpleflea.com/docs/migration/v1-to-v2>; rel="deprecation"

{
  "balance": 142.50,
  "currency": "USDT"
}

import httpx
import logging
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from typing import Optional

logger = logging.getLogger(__name__)

class DeprecationAwareClient:
    """
    HTTP client that detects and logs API deprecation warnings.
    Raises DeprecationWarning when sunset date is within 30 days.
    """

    def __init__(self, base_url: str, api_key: str,
                 warn_days_before_sunset: int = 30):
        self.base_url  = base_url
        self.api_key   = api_key
        self.warn_days = warn_days_before_sunset
        self._client   = httpx.Client(
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10.0,
        )

    def _check_deprecation(self, response: httpx.Response):
        deprecation = response.headers.get("Deprecation")
        sunset_str  = response.headers.get("Sunset")
        link        = response.headers.get("Link", "")

        if not deprecation:
            return

        msg = f"API endpoint {response.url} is deprecated."

        if sunset_str:
            try:
                sunset_dt = parsedate_to_datetime(sunset_str)
                now       = datetime.now(timezone.utc)
                days_left = (sunset_dt - now).days
                msg += f" Sunset in {days_left} days ({sunset_str})."

                if days_left <= self.warn_days:
                    import warnings
                    warnings.warn(
                        f"{msg} Migrate soon! See: {link}",
                        DeprecationWarning,
                        stacklevel=3,
                    )
            except Exception:
                pass

        logger.warning(msg + f" Migration: {link}")

    def get(self, path: str, **kwargs) -> httpx.Response:
        url  = f"{self.base_url}{path}"
        resp = self._client.get(url, **kwargs)
        self._check_deprecation(resp)
        return resp

    def post(self, path: str, **kwargs) -> httpx.Response:
        url  = f"{self.base_url}{path}"
        resp = self._client.post(url, **kwargs)
        self._check_deprecation(resp)
        return resp


# Usage
client = DeprecationAwareClient(
    base_url = "https://casino.purpleflea.com/api/v1",
    api_key  = "pf_live_<your_key>",
)
resp = client.get("/balance")
# If v1 is deprecated: logs warning + triggers DeprecationWarning
print(resp.json())

import httpx
import logging
from typing import Optional

logger = logging.getLogger(__name__)

KNOWN_VERSIONS = ["v3", "v2", "v1"]   # newest first

class VersionNegotiatingClient:
    """
    Automatically discovers the highest supported API version.
    Falls back to lower versions if the preferred version is unavailable.
    Caches the negotiated version for the session lifetime.
    """

    def __init__(self, base_url: str, api_key: str,
                 preferred: str = "v3",
                 fallback_order: list = None):
        self.base_url        = base_url.rstrip("/")
        self.api_key         = api_key
        self.preferred       = preferred
        self.fallback_order  = fallback_order or KNOWN_VERSIONS
        self._active_version: Optional[str] = None
        self._client         = httpx.Client(
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10.0,
        )

    def _probe_version(self, version: str) -> bool:
        """Check if `version` is supported by calling /health."""
        try:
            resp = self._client.get(
                f"{self.base_url}/{version}/health",
                timeout=5.0,
            )
            return resp.status_code == 200
        except httpx.RequestError:
            return False

    def negotiate(self) -> str:
        """Return the highest supported API version."""
        if self._active_version:
            return self._active_version

        # Try preferred first, then fall back in order
        candidates = [self.preferred] + [
            v for v in self.fallback_order if v != self.preferred
        ]
        for version in candidates:
            if self._probe_version(version):
                self._active_version = version
                logger.info("Negotiated API version: %s", version)
                if version != self.preferred:
                    logger.warning(
                        "Preferred version %s unavailable; using %s",
                        self.preferred, version
                    )
                return version

        raise RuntimeError("No supported API version found")

    def _url(self, path: str) -> str:
        return f"{self.base_url}/{self.negotiate()}{path}"

    def get(self, path: str, **kwargs) -> dict:
        resp = self._client.get(self._url(path), **kwargs)
        resp.raise_for_status()
        return resp.json()

    def post(self, path: str, body: dict = None, **kwargs) -> dict:
        resp = self._client.post(self._url(path), json=body, **kwargs)
        resp.raise_for_status()
        return resp.json()

# Example: agent starts up and negotiates the best available version
client = VersionNegotiatingClient(
    base_url  = "https://casino.purpleflea.com/api",
    api_key   = "pf_live_<your_key>",
    preferred = "v3",
)
balance = client.get("/balance")
print(f"Using API {client._active_version}: balance = {balance}")

import pytest
import httpx
from typing import Any

API_KEY  = "pf_live_<your_key>"
BASE_V2  = "https://casino.purpleflea.com/api/v2"

@pytest.fixture(scope="session")
def client():
    return httpx.Client(
        base_url=BASE_V2,
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=10.0,
    )

# ── Schema contracts ────────────────────────────────────────────────────

def assert_has_keys(data: dict, *keys: str):
    for k in keys:
        assert k in data, f"Missing key '{k}' in response: {data}"

def assert_type(value: Any, expected_type, field: str):
    assert isinstance(value, expected_type), (
        f"Field '{field}' expected {expected_type.__name__}, "
        f"got {type(value).__name__}"
    )

# ── Tests ───────────────────────────────────────────────────────────────

def test_balance_schema(client):
    resp = client.get("/balance")
    assert resp.status_code == 200
    data = resp.json()
    # Required v2 fields
    assert_has_keys(data, "balance", "currency")
    assert_type(data["balance"],  (int, float), "balance")
    assert_type(data["currency"], str,           "currency")

def test_balance_no_extra_required_fields(client):
    """Ensure we handle new fields gracefully (ignore unknowns)."""
    resp = client.get("/balance")
    data = resp.json()
    # Fields we depend on are present and correct type
    assert data["balance"] >= 0
    # We do NOT assert data.keys() == {...} — new fields are fine

def test_bet_schema(client):
    resp = client.post("/bet", json={"game": "dice", "amount": 0.01})
    assert resp.status_code in (200, 201, 402)   # OK, created, or insufficient funds
    if resp.status_code in (200, 201):
        data = resp.json()
        assert_has_keys(data, "result", "payout", "tx_id")

def test_version_header_present(client):
    resp = client.get("/balance")
    assert "X-Api-Version" in resp.headers or "x-api-version" in resp.headers

def test_deprecation_header_absent(client):
    """Fail CI if this version is already deprecated."""
    resp = client.get("/balance")
    assert "Deprecation" not in resp.headers, (
        "API version is deprecated! Update agent configuration."
    )

def test_error_schema(client):
    """Error responses must have 'error' and 'code' fields."""
    resp = client.post("/bet", json={"game": "invalid-game"})
    if resp.status_code >= 400:
        data = resp.json()
        assert_has_keys(data, "error")

# Run: pytest compat_tests.py -v

name: API Compatibility Check
on:
  schedule:
    - cron: '0 6 * * *'    # daily at 06:00 UTC
  push:
    branches: [main]

jobs:
  compat:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.12' }
      - run: pip install pytest httpx
      - run: pytest compat_tests.py -v
        env:
          PF_API_KEY: ${{ secrets.PF_API_KEY }}

# v1 balance response:
# { "amount": 142.50, "symbol": "USDT" }

# v2 balance response:
# { "balance": 142.50, "currency": "USDT", "balance_available": 132.50 }

# Migration shim — wraps v1 response in v2 shape
def normalize_balance(raw: dict, from_version: str = "v2") -> dict:
    if from_version == "v1":
        return {
            "balance":           raw.get("amount", 0),
            "currency":          raw.get("symbol", "USDT"),
            "balance_available": raw.get("amount", 0),
        }
    return raw   # v2+ already correct

# Use during transition period
client_v1 = PurpleFleatClient(
    api_key="pf_live_<your_key>", version="v1")
raw    = client_v1.balance()
normal = normalize_balance(raw, from_version="v1")
print(normal["balance"])   # works for both v1 and v2

def safe_get_balance(client) -> float:
    """
    Return available balance, defaulting gracefully on schema surprises.
    """
    try:
        data = client.balance()

        # Prefer most specific field; fall back to broader ones
        return float(
            data.get("balance_available")
            or data.get("balance")
            or data.get("amount")      # v1 compat
            or 0
        )
    except (KeyError, TypeError, ValueError) as exc:
        logger.warning("Balance parse error: %s — returning 0", exc)
        return 0.0


def safe_bet(client, game: str, amount: float) -> dict:
    """Place bet with full error handling for any API version."""
    try:
        result = client.bet(game=game, amount=amount)
        return {
            "success": True,
            "payout":  float(result.get("payout", 0)),
            "tx_id":   result.get("tx_id") or result.get("id", ""),
        }
    except httpx.HTTPStatusError as e:
        body = {}
        try:
            body = e.response.json()
        except Exception:
            pass
        return {
            "success": False,
            "error":   body.get("error", str(e)),
            "code":    body.get("code", e.response.status_code),
        }