lubes 0.1.0

Official Python SDK for Lubes

agentbase

Official Python SDK for AgentBase.

Installation

pip install agentbase

Quick Start

from agentbase import AgentBase

# Initialize with API key
client = AgentBase(
    api_key="your-api-key",
    org="your-org",
    project="your-project",
)

# List tables
tables = client.list_tables()
for table in tables:
    print(table.name)

# Execute a query
result = client.execute_query("SELECT * FROM users LIMIT 10")
for row in result.rows:
    print(row)

Authentication

API Key Authentication

client = AgentBase(
    api_key="ab_your_api_key_here",
    org="my-org",
    project="my-project",
)

Email/Password Authentication

client = AgentBase(api_url="https://api.agentbase.dev")

# Login
user, tokens = client.login(
    email="[e-mail]",
    password="password",
)

# Set org and project
client.set_org("my-org").set_project("my-project")

Async Support

import asyncio
from agentbase import AsyncAgentBase

async def main():
    async with AsyncAgentBase(
        api_key="your-api-key",
        org="my-org",
        project="my-project",
    ) as client:
        # List tables
        tables = await client.list_tables()

        # Execute query
        result = await client.execute_query("SELECT * FROM users")
        print(result.rows)

asyncio.run(main())

Database Operations

List Tables

tables = client.list_tables()
for table in tables:
    print(f"{table.name}: {table.row_count} rows")

Get Table Schema

schema = client.get_table_schema("users")
for column in schema.columns:
    print(f"{column.name}: {column.type}")

Execute Queries

# Simple query
result = client.execute_query("SELECT * FROM users WHERE active = true")

# Parameterized query
result = client.execute_query(
    "SELECT * FROM users WHERE email = $1",
    params=["[e-mail]"]
)

print(f"Found {result.row_count} rows")
for row in result.rows:
    print(row)

Database Branches

# List branches
branches = client.list_branches()

# Create branch
branch = client.create_branch(name="feature-branch")

# Delete branch
client.delete_branch("feature-branch")

Functions

List Functions

functions = client.list_functions()
for fn in functions:
    print(f"{fn.name} ({fn.runtime})")

Invoke Function

result = client.invoke_function("my-function", payload={"input": "data"})
print(result.result)
print(f"Execution time: {result.execution_time_ms}ms")

Function Logs

logs = client.get_function_logs("my-function", limit=50)
for log in logs:
    print(f"[{log.level}] {log.message}")

Deployments

# List deployments
deployments = client.list_deployments()

# Create deployment
deployment = client.create_deployment(description="New feature release")

# Rollback
client.rollback_deployment(deployment.id)

Environment Variables

# List env vars
env_vars = client.list_env_vars()

# Create env var
env_var = client.create_env_var(
    key="API_KEY",
    value="secret-value",
    is_secret=True,
)

# Reveal secret value
revealed = client.reveal_env_var(env_var.id)
print(revealed.value)

Storage

Buckets

# List buckets
buckets = client.list_buckets()

# Create bucket
bucket = client.create_bucket(name="my-bucket", is_public=False)

Objects

# List objects
objects, total, has_more = client.list_objects(bucket.id, prefix="uploads/")

# Get signed URL for download
signed_url = client.get_signed_url(
    bucket.id,
    key="uploads/file.pdf",
    expires_in=3600,
)
print(signed_url.url)

# Delete object
client.delete_object(bucket.id, "uploads/file.pdf")

Webhooks

# Create webhook
webhook = client.create_webhook(
    name="My Webhook",
    url="https://example.com/webhook",
    events=["deployment.created", "deployment.completed"],
)

# Get deliveries
deliveries, total = client.get_webhook_deliveries(webhook.id)

Organization Management

# List organizations
orgs = client.list_organizations()

# Create organization
org = client.create_organization(name="My Org", slug="my-org")

# Invite member
member = client.invite_member(email="[e-mail]", role="member")

Error Handling

from agentbase import AgentBase, AgentBaseError

try:
    result = client.execute_query("INVALID SQL")
except AgentBaseError as e:
    print(f"Error: {e.code} - {e.message}")
    if e.details:
        print(f"Details: {e.details}")

Configuration Options

client = AgentBase(
    api_url="https://api.agentbase.dev",  # API URL (default)
    api_key="your-api-key",               # API key authentication
    access_token="jwt-token",             # JWT token authentication
    org="your-org",                       # Organization slug
    project="your-project",               # Project slug
    timeout=30.0,                         # Request timeout in seconds (default: 30)
)

Context Manager

# Synchronous
with AgentBase(api_key="key", org="org", project="project") as client:
    tables = client.list_tables()

# Asynchronous
async with AsyncAgentBase(api_key="key", org="org", project="project") as client:
    tables = await client.list_tables()

Type Hints

Full type hints are provided via Pydantic models:

from agentbase import (
    User,
    Organization,
    Project,
    TableSchema,
    QueryResult,
    ServerlessFunction,
    Deployment,
    EnvVar,
    Bucket,
    StorageObject,
    Webhook,
)

License

MIT