contextbase 0.0.3

Official Python SDK for Contextbase API

Contextbase Python SDK

The official Python SDK for Contextbase

Installation

pip install contextbase

Quick start

Setup

First, set your API key as an environment variable:

export CONTEXTBASE_API_KEY="your-api-key-here"

Or pass it directly when initializing the client:

from contextbase import Contextbase

client = Contextbase(api_key="your-api-key-here")

Basic usage

Publishing JSON data

from contextbase import Contextbase

cb = Contextbase()

# Publish JSON data
response = cb.publish(
    context_name="user_analytics",
    body={"action": "login", "timestamp": "2024-01-15T10:30:00Z"},
    scopes={"user_id": 123}
)

if response.ok:
    print("Data published successfully!")
    print(f"Response: {response.json}")
else:
    print(f"Error: {response.error.message}")

Publishing files

from contextbase import Contextbase, ContextbaseFile

cb = Contextbase()

# Method 1: Use the convenience method
response = cb.publish_file(
    context_name="reports",
    file_path="report.pdf"
)

# Method 2: Create file from path
file = ContextbaseFile.from_path("path/to/your/document.pdf")
response = cb.publish(
    context_name="reports",
    file=file
)

# Method 3: Create from data content
content = "Daily report: All systems operational!"
file = ContextbaseFile.from_data(content, "daily-report.txt")
response = cb.publish(
    context_name="daily_reports",
    file=file
)

# Method 4: Create from binary data
with open("image.png", "rb") as f:
    binary_data = f.read()
file = ContextbaseFile.from_data(binary_data, "screenshot.png")
response = cb.publish(
    context_name="screenshots",
    file=file
)

Resolving context

# Basic query
response = client.resolve_context("my-app")

# Query with search term
response = client.resolve_context(
    context_name="my-app",
    query="user login events"
)

# Query with scopes
response = client.resolve_context(
    context_name="my-app",
    scopes={"environment": "production", "date_range": "last_week"}
)

if response.ok:
    print(f"Resolved context response: {response.json}")

Using the decorator

The @publish decorator automatically publishes function results to Contextbase:

For JSON data

from contextbase import publish

@publish(context_name="ml-models")
def predict_user_behavior(user_data):
    # Your ML logic here
    prediction = {"user_id": user_data["id"], "likely_to_churn": 0.23}
    return prediction

# Function runs normally, and result is automatically published
result = predict_user_behavior({"id": 123, "activity": "low"})

For file output

# Automatically upload function output as a file
@publish(
    context_name="daily_reports", 
    as_file=True,
    file_name="summary.txt"
)
def generate_daily_report():
    return "Daily Summary: All systems operational!"

# Content is automatically converted to ContextbaseFile and uploaded
report = generate_daily_report()

# Works with binary data too
@publish(
    context_name="generated_charts",
    as_file=True,
    file_name="chart.png"
)
def create_chart():
    # Return binary PNG data
    return generate_png_bytes()

Decorator with error handling

# Raise exceptions on publish failures
@publish(
    context_name="financial_calculations", 
    scopes=lambda result: {"category": "critical" if result["risk_score"] > 0.70 else "normal"},
    raise_on_error=True
)
def calculate_risk_score(portfolio):
    return {"risk_score": 0.75, "confidence": 0.92}

# Silently continue on publish failures (default)
@publish(
    context_name="user_events", 
    raise_on_error=False
)
def track_user_action(user_id, action):
    return {"user_id": user_id, "action": action}

Decorator with scopes

@publish(
    context_name="system_metrics",
    scopes={"environment": "production", "service": "api"}
)
def collect_metrics():
    return {
        "cpu_usage": 45.2,
        "memory_usage": 67.8,
        "timestamp": "2024-01-15T10:30:00Z"
    }

Advanced usage

Error handling

from contextbase import Contextbase, ContextbaseError

client = Contextbase()

try:
    response = client.publish("context", "component", body={"data": "value"})
    response.raise_for_status()  # Raises ContextbaseError if response failed
    print("Success!")
except ContextbaseError as e:
    print(f"API Error: {e.message}")
    print(f"Status Code: {e.status_code}")
    for error in e.errors:
        print(f"  - {error}")
except Exception as e:
    print(f"Unexpected error: {e}")

Response object methods

response = client.publish("context", "component", body={"data": "value"})

# Check success
if response.ok:  # or response.is_success
    print("Request successful")

# Access response data
data = response.json          # Parsed JSON response
text = response.text          # Raw response text  
headers = response.headers    # Response headers dict

# Dict-like access
value = response.get("key", "default")
if "field" in response:
    field_value = response["field"]

# Error information
if not response.ok:
    error = response.error
    print(f"Error: {error.message}")
    print(f"Details: {error.errors}")

File upload reference

ContextbaseFile class

The ContextbaseFile class provides an interface for file publishing:

from contextbase import ContextbaseFile

# Create from file path
file = ContextbaseFile.from_path("document.pdf")

# Create from string content
file = ContextbaseFile.from_data("Hello World", "greeting.txt")

# Create from binary content  
with open("image.png", "rb") as f:
    binary_data = f.read()
file = ContextbaseFile.from_data(binary_data, "image.png")

# Create with explicit MIME type
file = ContextbaseFile.from_data(
    content="Custom content",
    name="data.custom",
    mime_type="application/custom"
)

# Direct constructor
import base64
file = ContextbaseFile(
    name="file.txt",
    mime_type="text/plain",
    base64_content=base64.b64encode(b"content").decode()
)

# Access file properties
print(file.name)           # "file.txt"
print(file.mime_type)      # "text/plain"
print(file.get_content())  # b"content"
print(file.get_size())     # 7

Supported file types

See documentation

Error reference

ContextbaseError

Raised when the API returns an error response:

try:
    response = client.publish("context", "component", body={})
    response.raise_for_status()
except ContextbaseError as e:
    print(f"Status: {e.status_code}")     # HTTP status code
    print(f"Message: {e.message}")        # Error message
    print(f"Details: {e.errors}")         # List of detailed errors

Development

Running tests

# Install development dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run tests with coverage
pytest --cov=contextbase --cov-report=html

# Run specific test file
pytest tests/test_file_upload.py -v

Support

• Documentation: https://docs.contextbase.dev
• Issues: GitHub Issues
• Email: [e-mail]

License

This project is licensed under the MIT License - see the LICENSE file for details.