buddo-py 0.1.0a1

Python SDK for the Buddo loyalty platform API

buddo-py

Async Python SDK for the Buddo loyalty platform.

Buddo is a Bitcoin-backed loyalty system for operators — games, apps, and websites. Operators earn ad-share revenue when their users view ads; users earn Buddo points redeemable for in-app value. This SDK wraps the Buddo REST API with typed responses, automatic token refresh, retry/backoff, and rate-limit compliance built in.

Status: Alpha (v0.1.0a1) — API is stable within the 0.1.x line but may change before v1.0. Pin your dependency to a minor version in production.

---

Installation

pip install buddo-py

With optional extras:

# Trading bot extras (pandas, numpy, tenacity for strategy work)
pip install "buddo-py[trading]"

# Development tools (pytest, mypy, ruff, respx)
pip install "buddo-py[dev]"

Recommended: use a virtual environment

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install buddo-py

Verify the install:

python -c "import buddo; print(buddo.__version__)"
# 0.1.0a1

Requires Python 3.9+. Core dependencies (httpx, pydantic) are installed automatically.

---

Quick Example

import asyncio
import os
from buddo import BuddoClient

async def main() -> None:
    async with BuddoClient(api_key=os.environ["BUDDO_API_KEY"]) as client:
        # Check operator treasury balance
        balance = await client.users.get_balance()
        print(f"Treasury: {balance.buddo_points} pts  (tier: {balance.tier})")

        # Award 100 points to a user (operator-scoped, requires points:award scope)
        txn = await client.users.award_points(
            user_id="usr_abc123",
            amount=100,
            reason="ad_view",
            idempotency_key="award_001",   # Required — enables safe retries
        )
        print(f"Awarded {txn.amount} pts — transaction: {txn.id}")

asyncio.run(main())

See docs/quickstart.md for the full step-by-step guide.

---

Configuration

Getting an API Key

1. Sign in to the Buddo operator dashboard
2. Navigate to Settings → API Keys
3. Create a key — it will be prefixed bdk_live_ (production) or bdk_test_ (sandbox)

Prefix	Environment	Use
bdk_live_	Production	Real transactions, real points
bdk_test_	Sandbox	Safe for development, no real treasury impact

Setting the Environment Variable

# macOS / Linux
export BUDDO_API_KEY="bdk_test_your_key_here"

# Windows PowerShell
$env:BUDDO_API_KEY = "bdk_test_your_key_here"

For local development, use a .env file with python-dotenv:

# .env — never commit to version control
BUDDO_API_KEY=bdk_test_your_key_here

from dotenv import load_dotenv
load_dotenv()

Production: store the key in a secrets manager (AWS Secrets Manager, HashiCorp Vault, etc.). Never commit API keys to version control — add .env to your .gitignore.

---

Resources

BuddoClient exposes six resource objects. Each resource maps directly to an API endpoint group:

Resource	Description
client.auth	OAuth2 token management: login, refresh, revoke, client credentials flow
client.users	User profiles, Buddo point balances, transaction history, and point awards
client.social	Friends list, friend requests, presence status, and lobby chat
client.games	Game catalogue, session creation, and session settlement
client.ads	Ad campaign management, impression recording, and campaign analytics
client.operator	Operator stats, registered app management, and monthly billing

Key Methods

client.users

• get_profile() → UserProfile — authenticated user's profile
• get_balance() → UserBalance — Buddo points balance and loyalty tier
• get_points_history(limit, cursor) → PagedResult[PointsTransaction]
• award_points(user_id, amount, reason, idempotency_key) → PointsTransaction
• get(user_id) → UserProfile — look up any user (operator scope)

client.ads

• list_campaigns(status) → List[Campaign]
• create_campaign(name, budget, cpm, start_date, end_date, ...) → Campaign
• record_impression(campaign_id, user_id, ad_type, duration_seconds, idempotency_key) → Impression
• get_campaign_stats(campaign_id) → CampaignStats

client.operator

• get_stats() → OperatorStats
• create_app(name, redirect_uris, scopes) → OperatorApp
• rotate_secret(app_id) → OperatorApp
• get_billing(month) → BillingRecord

---

Error Handling

All SDK errors subclass BuddoError. Catch specific subclasses for fine-grained control, or catch the base class as a fallback:

import asyncio
import os
from buddo import BuddoClient
from buddo.exceptions import (
    BuddoRateLimitError,
    BuddoInsufficientBalanceError,
    BuddoNotFoundError,
    BuddoConflictError,
    BuddoAuthError,
    BuddoError,
)


async def award_points_safe(
    client: BuddoClient,
    user_id: str,
    amount: int,
    idempotency_key: str,
) -> None:
    try:
        txn = await client.users.award_points(
            user_id=user_id,
            amount=amount,
            reason="ad_view",
            idempotency_key=idempotency_key,
        )
        print(f"Awarded {txn.amount} pts — txn: {txn.id}, balance after: {txn.balance_after}")

    except BuddoRateLimitError as exc:
        # Raised only after the SDK's built-in retries (default: 3) are exhausted.
        # exc.retry_after: seconds to wait before the next attempt is safe.
        print(f"Rate limited — retry after {exc.retry_after}s")
        await asyncio.sleep(exc.retry_after)

    except BuddoInsufficientBalanceError as exc:
        # exc.required: points needed; exc.available: current treasury balance.
        print(f"Treasury low: need {exc.required} pts, have {exc.available} pts")

    except BuddoNotFoundError:
        print(f"User {user_id!r} does not exist")

    except BuddoConflictError:
        # Same idempotency_key used with a *different* payload — this is a bug.
        # Same key + same payload → original response returned (no exception).
        print("Idempotency key conflict — check for duplicate requests with different params")

    except BuddoAuthError:
        print("Authentication failed — check your API key or re-authenticate")
        raise  # Re-raise; the bot should not continue with a bad credential

    except BuddoError as exc:
        # Catch-all: includes BuddoServerError, BuddoForbiddenError, etc.
        print(f"SDK error [{exc.status_code}] {exc.error_code}: {exc}")
        if exc.retryable:
            print("  (transient — safe to retry after a short delay)")

Exception hierarchy:

BuddoError
├── BuddoAuthError          401 — invalid/expired token
├── BuddoForbiddenError     403 — valid token, insufficient scope
├── BuddoNotFoundError      404 — resource not found
├── BuddoConflictError      409 — duplicate idempotency key with different payload
├── BuddoRateLimitError     429 — rate limit (.retry_after seconds)
├── BuddoServerError        5xx — server error (502/503/504 are retryable)
├── BuddoValidationError    422 — request validation failed (.field_errors dict)
├── BuddoInsufficientBalanceError  402 — not enough points (.required, .available)
├── BuddoTimeoutError       Network timeout (retryable)
└── BuddoConnectionError    Network connection error (retryable)

---

Async Usage

The SDK is async-only: every resource method is a coroutine. The HTTP layer uses httpx.AsyncClient for non-blocking I/O, making it suitable for high-throughput operator bots that need to award points or record impressions without blocking the event loop.

Recommended: async context manager

async with BuddoClient(api_key=os.environ["BUDDO_API_KEY"]) as client:
    balance = await client.users.get_balance()
    # Connection pool is released automatically on exit

For long-lived bots: manual lifecycle

client = BuddoClient(api_key=os.environ["BUDDO_API_KEY"])
await client.open()

try:
    while True:
        await run_bot_cycle(client)
        await asyncio.sleep(60)
finally:
    await client.aclose()  # Always close — releases the connection pool

Calling from synchronous code:

import asyncio

result = asyncio.run(some_async_function())

---

Pagination

List endpoints that may return many items use cursor-based pagination via PagedResult[T]:

from buddo.models import PagedResult
from buddo.models.user import PointsTransaction

async def fetch_all_history(client: BuddoClient) -> list[PointsTransaction]:
    all_txns: list[PointsTransaction] = []
    cursor: str | None = None

    while True:
        page = await client.users.get_points_history(limit=50, cursor=cursor)
        all_txns.extend(page.data)

        if not page.has_next:
            break
        cursor = page.cursor

    return all_txns

PagedResult fields:

• .data — list of items on this page
• .cursor — opaque string to pass as cursor= on the next request; None when no more pages
• .has_next — True when more pages remain
• .total — total count when the server provides it (not always available)

---

Trading Bot Example

For bots that manage Buddo balances alongside trading activity:

import asyncio
import logging
import os
from buddo import BuddoClient
from buddo.exceptions import BuddoRateLimitError, BuddoInsufficientBalanceError, BuddoError

logger = logging.getLogger(__name__)

MIN_TREASURY_BALANCE = 1_000  # Don't trade if treasury is critically low


async def run_trading_cycle(client: BuddoClient, user_id: str) -> None:
    """Check balance, execute a trade, then award loyalty points."""
    try:
        balance = await client.users.get_balance()
        if balance.buddo_points < MIN_TREASURY_BALANCE:
            logger.warning(
                "Treasury low (%d pts) — skipping award this cycle",
                balance.buddo_points,
            )
            return

        # --- your trading logic here ---
        trade_session_id = "session_example_001"
        logger.info("Treasury OK (%d pts) — executing trade", balance.buddo_points)

        # Award points as loyalty reward for the trade
        txn = await client.users.award_points(
            user_id=user_id,
            amount=50,
            reason="trade_bonus",
            idempotency_key=f"trade_{trade_session_id}_{user_id}",
        )
        logger.info("Loyalty award sent — txn: %s, balance after: %d", txn.id, txn.balance_after)

    except BuddoRateLimitError as exc:
        logger.warning("Rate limited — sleeping %ds", exc.retry_after)
        await asyncio.sleep(exc.retry_after)

    except BuddoInsufficientBalanceError as exc:
        logger.error(
            "Treasury exhausted (need %d, have %d) — top up via operator dashboard",
            exc.required,
            exc.available,
        )

    except BuddoError as exc:
        logger.error("Buddo API error [%s]: %s", exc.status_code, exc)


async def main() -> None:
    client = BuddoClient(api_key=os.environ["BUDDO_API_KEY"])
    await client.open()
    try:
        while True:
            await run_trading_cycle(client, user_id="usr_abc123")
            await asyncio.sleep(30)
    finally:
        await client.aclose()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

---

Advanced: OAuth2 Client Credentials

For machine-to-machine flows with scoped tokens:

async with BuddoClient(
    client_id=os.environ["BUDDO_CLIENT_ID"],
    client_secret=os.environ["BUDDO_CLIENT_SECRET"],
) as client:
    token = await client.auth.client_credentials(scope="operator:read operator:write")
    # The client uses this token for subsequent requests
    stats = await client.operator.get_stats()

---

Links

• Quickstart Guide — award points in 5 minutes
• AGENTS.md — LLM agent integration reference
• Full API Reference
• OpenAPI Spec
• Buddo Platform
• Issue Tracker

---

Contributing

Contributions are welcome. The project uses hatchling as its build backend and requires Python 3.9+.

# 1. Fork and clone
git clone https://github.com/garybots/buddo-py.git
cd buddo-py

# 2. Install with dev extras
pip install -e ".[dev]"

# 3. Run the test suite
pytest

# 4. Type-check
mypy buddo/

# 5. Lint
ruff check buddo/

Tests use pytest-asyncio with asyncio_mode = "auto" — no @pytest.mark.asyncio decorator needed. HTTP calls are mocked with respx; no live API access required to run the test suite.

Please open an issue before submitting a large PR. For bug reports, include the SDK version (buddo.__version__), Python version, and a minimal reproduction.

---

License

MIT License — see LICENSE for details.