business-use 0.4.0

Lightweight Python SDK for Business-Use event tracking and assertions

Business-Use Python SDK

A lightweight, production-ready Python SDK for tracking business events and assertions in your applications.

Features

• Simple API: Main function ensure() with convenience helpers act() and assert_()
• Non-blocking: Events are batched and sent asynchronously in the background
• Never fails: All errors are handled internally - your application never crashes
• Thread-safe: Safe to use from multiple threads concurrently
• Minimal dependencies: Only httpx and pydantic
• Context-aware: Filters and validators can access upstream dependency data

Installation

pip install business-use

Or with uv:

uv add business-use

Quick Start

from business_use import initialize, ensure

# Initialize with API key
initialize(api_key="your-api-key")

# Track a business action (no validator)
ensure(
    id="user_signup",
    flow="onboarding",
    run_id="user_12345",
    data={"email": "[e-mail]", "plan": "premium"}
)

# Track a business assertion (with validator)
def validate_payment(data, ctx):
    """Validate payment and check upstream dependencies."""
    # ctx["deps"] contains upstream dependency events
    # Each dep: {"flow": str, "id": str, "data": dict}
    has_cart = any(dep["id"] == "cart_created" for dep in ctx["deps"])
    return data["amount"] > 0 and has_cart

ensure(
    id="payment_valid",
    flow="checkout",
    run_id="order_67890",
    data={"amount": 99.99, "currency": "USD"},
    dep_ids=["cart_created"],
    validator=validate_payment  # Creates "assert" node
)

# Convenience wrappers (optional)
from business_use import act, assert_

# act() is ensure() without validator
act(id="user_signup", flow="onboarding", run_id="u123", data={...})

# assert_() is ensure() with validator
assert_(id="validation", flow="checkout", run_id="o123", data={...}, validator=fn)

API Reference

initialize()

Initialize the SDK before using ensure(). This validates the connection to the backend and starts the background batch processor.

Parameters:

• api_key (str, optional): API key for authentication (defaults to BUSINESS_USE_API_KEY env var)
• url (str, optional): Backend API URL (defaults to BUSINESS_USE_URL env var or http://localhost:13370)
• batch_size (int, optional): Number of events per batch (default: 100)
• batch_interval (int, optional): Flush interval in seconds (default: 5)
• max_queue_size (int, optional): Max queue size (default: batch_size * 10)

Examples:

# With explicit parameters
initialize(
    api_key="your-api-key",
    url="https://api.example.com",
    batch_size=50,
    batch_interval=10
)

# With environment variables
# Set BUSINESS_USE_API_KEY and BUSINESS_USE_URL in your environment
initialize()

# Mix of both (parameters override env vars)
initialize(api_key="your-api-key")  # URL from BUSINESS_USE_URL or default

ensure()

Main function to track business events. Type is auto-determined by the presence of a validator.

Parameters:

• id (str, required): Unique node/event identifier
• flow (str, required): Flow identifier
• run_id (str | callable, required): Run identifier (or lambda returning string)
• data (dict, required): Event data payload
• filter (callable, optional): Filter function (data, ctx) -> bool evaluated on backend
• dep_ids (list[str] | callable, optional): Dependency node IDs
• validator (callable, optional): Validation function (data, ctx) -> bool executed on backend
• description (str, optional): Human-readable description
• conditions (list[NodeCondition] | callable, optional): Optional conditions (e.g., timeout constraints)
• additional_meta (dict, optional): Optional additional metadata

Type determination:

• If validator is provided → creates an "assert" node
• If validator is None → creates an "act" node

Example:

# Action node (no validator)
ensure(
    id="payment_processed",
    flow="checkout",
    run_id="run_12345",
    data={"amount": 100, "currency": "USD"},
    dep_ids=["cart_created"],
    description="Payment processed successfully"
)

# Assertion node (with validator)
def validate_total(data, ctx):
    items_total = sum(dep["data"]["price"] for dep in ctx["deps"])
    return data["total"] == items_total

ensure(
    id="order_total_valid",
    flow="checkout",
    run_id="run_12345",
    data={"total": 150},
    dep_ids=["item_added"],
    validator=validate_total,
    description="Order total validation"
)

act()

Convenience wrapper around ensure() without a validator. Creates an "act" node.

Parameters:

• id (str, required): Unique node/event identifier
• flow (str, required): Flow identifier
• run_id (str | callable, required): Run identifier (or lambda returning string)
• data (dict, required): Event data payload
• filter (callable, optional): Filter function (data, ctx) -> bool evaluated on backend
• dep_ids (list[str] | callable, optional): Dependency node IDs
• description (str, optional): Human-readable description

Filter Function Signature:

def my_filter(data: dict, ctx: dict) -> bool:
    """Filter function executed on the backend.

    Args:
        data: Current event data
        ctx: Context with upstream dependencies:
             ctx["deps"] = [{"flow": str, "id": str, "data": dict}, ...]

    Returns:
        bool: True to include event, False to filter it out
    """
    pass

Example:

# Simple usage
act(
    id="payment_processed",
    flow="checkout",
    run_id="run_12345",
    data={"amount": 100, "currency": "USD"}
)

# With dependencies
act(
    id="order_completed",
    flow="checkout",
    run_id="run_12345",
    data={"order_id": "ord_123", "total": 150},
    dep_ids=["payment_processed", "inventory_reserved"],
    description="Order completed successfully"
)

# Using lambdas for dynamic values
act(
    id="api_request",
    flow="integration",
    run_id=lambda: get_current_trace_id(),
    data={"endpoint": "/users", "method": "POST"},
    dep_ids=lambda: get_upstream_dependencies()
)

# With filter based on upstream dependencies
def check_prerequisites(data, ctx):
    """Only process if all upstream tasks are approved."""
    return all(dep["data"].get("status") == "approved" for dep in ctx["deps"])

act(
    id="order_completed",
    flow="checkout",
    run_id="run_12345",
    data={"order_id": "ord_123"},
    dep_ids=["payment_processed", "inventory_reserved"],
    filter=check_prerequisites,  # Evaluated on backend with ctx
    description="Order completed after all prerequisites"
)

assert_()

Convenience wrapper around ensure() with a validator. Creates an "assert" node.

Named assert_ (with underscore) to avoid conflict with Python's built-in assert keyword.

Parameters:

• id (str, required): Unique node/event identifier
• flow (str, required): Flow identifier
• run_id (str | callable, required): Run identifier (or lambda returning string)
• data (dict, required): Event data payload
• filter (callable, optional): Filter function (data, ctx) -> bool evaluated on backend
• dep_ids (list[str] | callable, optional): Dependency node IDs
• validator (callable, optional): Validation function (data, ctx) -> bool executed on backend
• description (str, optional): Human-readable description

Validator Function Signature:

def my_validator(data: dict, ctx: dict) -> bool:
    """Validator function executed on the backend.

    Args:
        data: Current event data
        ctx: Context with upstream dependencies:
             ctx["deps"] = [{"flow": str, "id": str, "data": dict}, ...]

    Returns:
        bool: True if validation passes, False if it fails
    """
    pass

Example:

# Simple assertion
assert_(
    id="order_total_positive",
    flow="checkout",
    run_id="run_12345",
    data={"total": 150}
)

# With validator accessing upstream dependencies
def validate_order(data, ctx):
    """Validate order total matches sum of items from upstream events.

    Args:
        data: Current order data
        ctx: Context with upstream item events in ctx["deps"]
    """
    # Calculate total from upstream item_added events
    items_total = sum(
        dep["data"]["price"]
        for dep in ctx["deps"]
        if dep["id"] == "item_added"
    )

    # Verify order total matches
    return data["total"] == items_total

assert_(
    id="order_total_matches",
    flow="checkout",
    run_id="run_12345",
    data={"total": 150},
    dep_ids=["item_added"],  # Multiple item_added events
    validator=validate_order,
    description="Order total matches sum of item prices"
)

# Validator with complex logic
def validate_payment_and_inventory(data, ctx):
    """Validate both payment and inventory are ready."""
    payment_approved = any(
        dep["id"] == "payment_processed" and dep["data"]["status"] == "approved"
        for dep in ctx["deps"]
    )
    inventory_reserved = any(
        dep["id"] == "inventory_reserved" and dep["data"]["reserved"] == True
        for dep in ctx["deps"]
    )
    return payment_approved and inventory_reserved and data["ready_to_ship"]

assert_(
    id="order_ready",
    flow="checkout",
    run_id="run_12345",
    data={"ready_to_ship": True},
    dep_ids=["payment_processed", "inventory_reserved"],
    validator=validate_payment_and_inventory
)

shutdown()

Gracefully shutdown the SDK and flush remaining events (optional - auto-shuts down on exit).

Parameters:

• timeout (float, optional): Max time to wait for shutdown in seconds (default: 5.0)

Example:

from business_use import shutdown

# At application shutdown
shutdown(timeout=10.0)

How It Works

Batching & Ingestion

The SDK uses a background worker thread to batch and send events:

1. Events are added to an in-memory queue (thread-safe)
2. Background thread collects events into batches
3. Batches are sent when:
   • Batch size reaches batch_size (default: 100), OR
   • batch_interval seconds elapse (default: 5)
4. On program exit, remaining events are flushed (best-effort)

Error Handling

The SDK never raises exceptions. All errors are caught and logged internally:

• Network errors: Logged, batch dropped
• Queue overflow: Oldest events dropped, logged
• Invalid parameters: Logged, no-op
• Not initialized: Silent no-op

Lambda Serialization

Callable parameters (run_id, dep_ids, filter, validator) are serialized as Python source code and executed on the backend.

Important Notes:

• Filters are evaluated on the backend with access to upstream dependencies via ctx
• Validators are executed on the backend with the same ctx structure
• Both filter and validator receive: (data: dict, ctx: dict) -> bool
• ctx["deps"] contains all upstream dependency events as a list of {"flow": str, "id": str, "data": dict}
• Only use simple lambdas or functions defined in the same file
• External references may fail serialization

Requirements

• Python >= 3.11
• httpx >= 0.27.0
• pydantic >= 2.0.0

Architecture

See SDK_ARCHITECTURE.md for detailed architecture documentation.

Development & Local Setup

Prerequisites

• Python >= 3.11
• uv (recommended) or pip

Setup

1. Clone the repository:

   git clone https://github.com/desplega-ai/business-use.git
   cd business-use/sdk-py
   

2. Install dependencies:

   # With uv (recommended)
   uv sync
   
   # Or with pip
   pip install -e ".[dev]"
   

3. Run the backend locally:

   The SDK requires the Business-Use backend API running locally:

   # In a separate terminal, navigate to the core directory
   cd ../core
   
   # Install core dependencies
   uv sync
   
   # Run the backend
   uv run cli serve --reload
   

   The API will be available at http://localhost:13370

Running Tests

# Run all tests
uv run pytest

# Run with verbose output
uv run pytest -v

# Run specific test file
uv run pytest tests/test_serialization.py

# Run with coverage
uv run pytest --cov=business_use --cov-report=html

Running the Example

Once the backend is running:

uv run python example.py

You should see:

• SDK initialization logs
• Events being tracked
• Batches being sent every 5 seconds
• Graceful shutdown

Project Structure

sdk-py/
├── src/business_use/
│   ├── __init__.py          # Public API exports
│   ├── client.py            # Main SDK (initialize, act, assert_)
│   ├── batch.py             # Background batching worker
│   └── models.py            # Pydantic models
├── tests/
│   ├── test_client.py       # Client behavior tests
│   └── test_serialization.py # Lambda serialization tests
├── example.py               # Usage example
├── pyproject.toml           # Project configuration
└── README.md                # This file

Development Workflow

1. Make changes to the SDK code in src/business_use/

2. Run tests to ensure nothing breaks:

   uv run pytest
   

3. Test manually with the example:

   # Terminal 1: Run backend
   cd ../core && uv run uvicorn src.api.api:app --port 13370
   
   # Terminal 2: Run example
   uv run python example.py
   

4. Check code quality (optional):

   # Format code
   uv run ruff format src/ tests/
   
   # Lint code
   uv run ruff check src/ tests/
   
   # Type check
   uv run mypy src/
   

Common Development Tasks

Add a new dependency

# Add runtime dependency
uv add package-name

# Add dev dependency
uv add --dev package-name

Debug SDK behavior

Enable debug logging in your code:

import logging

# Set SDK logger to debug level
logging.getLogger("business-use").setLevel(logging.DEBUG)

# Or configure all logging
logging.basicConfig(
    level=logging.DEBUG,
    format="[%(name)s] [%(levelname)s] %(message)s"
)

Test with different batch settings

Modify example.py to test different configurations:

initialize(
    api_key="your-api-key",
    batch_size=5,      # Small batch for testing
    batch_interval=1,  # Flush every second
)

Simulate backend failures

Stop the backend and observe SDK behavior:

• Connection check should fail gracefully
• SDK should enter no-op mode
• No exceptions should be raised

Environment Variables

The SDK supports configuration via environment variables. This is the recommended approach for production deployments:

# Required: Set API key
export BUSINESS_USE_API_KEY="your-api-key"

# Optional: Set custom backend URL (defaults to http://localhost:13370)
export BUSINESS_USE_URL="https://api.desplega.ai"

Then simply initialize without parameters:

from business_use import initialize

# Automatically uses BUSINESS_USE_API_KEY and BUSINESS_USE_URL from environment
initialize()

Benefits:

• ✅ Keep secrets out of code
• ✅ Easy configuration per environment (dev/staging/prod)
• ✅ Works with Docker, Kubernetes, etc.
• ✅ Parameters still override env vars when needed

Troubleshooting

Issue: SDK not sending events

1. Check backend is running:

   curl http://localhost:13370/health
   # Should return: {"status":"success","message":"API is healthy"}
   

2. Enable debug logging to see batch processing:

   logging.getLogger("business-use").setLevel(logging.DEBUG)
   

3. Check API key is correct (default: check backend config)

Issue: Tests failing

1. Ensure dependencies are up-to-date:

   uv sync
   

2. Run tests with verbose output:

   uv run pytest -v -s
   

3. Check if backend is interfering (tests should work without backend)

Issue: Import errors

Make sure you've installed the package in editable mode:

uv sync  # Reinstalls in editable mode

Backend API Endpoints

When running locally, the backend exposes:

• GET /health - Health check (no auth required)
• GET /v1/status - Verify API key
• POST /v1/events-batch - Batch event ingestion
• GET /v1/events - Query events
• GET /v1/nodes - Query nodes

Contributing

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes
4. Run tests (uv run pytest)
5. Commit your changes (git commit -m 'Add amazing feature')
6. Push to the branch (git push origin feature/amazing-feature)
7. Open a Pull Request

Best Practices

• Always run tests before committing
• Add tests for new features
• Keep dependencies minimal - avoid adding unnecessary packages
• Use type hints for better IDE support and maintainability
• Log errors instead of raising exceptions in SDK code
• Never block the main thread - all network I/O happens in background

License

MIT