Python SDK API Reference

pip install buena-sdk

import buena_sdk
from buena_sdk.api.default_api import DefaultApi
from buena_sdk.rest import ApiException

# Configure the client
configuration = buena_sdk.Configuration(
    host="https://api.buena.ai/api/v2"
)
configuration.api_key['ApiKeyAuth'] = 'your-api-key'

# Create API client
with buena_sdk.ApiClient(configuration) as api_client:
    api_instance = DefaultApi(api_client)

def list_leads(
    limit: Optional[int] = None,      # Number of leads (default: 50, max: 100)
    offset: Optional[int] = None,     # Pagination offset (default: 0)
    sort: Optional[str] = None,       # Sort field: 'created_at', 'updated_at', 'first_name', 'last_name'
    order: Optional[str] = None,      # Sort order: 'asc' or 'desc' (default: 'desc')
    search: Optional[str] = None,     # Search term for name/email
    company: Optional[str] = None,    # Filter by company name
    status: Optional[str] = None      # Filter by status: 'active', 'inactive', 'contacted'
) -> ListLeads200Response

class ListLeads200Response:
    data: List[Lead]           # List of lead objects
    total: int                 # Total number of leads matching filters
    limit: int                 # Applied limit
    offset: int                # Applied offset

class Lead:
    id: str                    # Unique lead identifier
    first_name: str            # Lead's first name
    last_name: str             # Lead's last name
    email: str                 # Email address
    company: Optional[str]     # Company name
    phone: Optional[str]       # Phone number
    linkedin_url: Optional[str] # LinkedIn profile URL
    job_title: Optional[str]   # Job title
    location: Optional[str]    # Location/city
    notes: Optional[str]       # Additional notes
    tags: Optional[List[str]]  # Array of tags
    custom_fields: Optional[Dict[str, Any]]  # Custom field data
    status: str                # Lead status
    created_at: datetime       # Creation timestamp
    updated_at: datetime       # Last update timestamp

try:
    # List first 25 leads, sorted by creation date
    response = api_instance.list_leads(
        limit=25,
        sort='created_at',
        order='desc'
    )

    print(f"Found {response.total} leads")
    for lead in response.data:
        print(f"{lead.first_name} {lead.last_name} - {lead.email}")
        if lead.company:
            print(f"  Company: {lead.company}")

    # Search for leads
    search_results = api_instance.list_leads(
        search="john",
        company="example"
    )
    print(f"Search found {len(search_results.data)} leads")

except ApiException as e:
    print(f"Exception when calling list_leads: {e}")

def create_lead(
    create_lead_request: CreateLeadRequest
) -> Lead

class CreateLeadRequest:
    first_name: str                    # Required: Lead's first name
    last_name: str                     # Required: Lead's last name
    email: str                         # Required: Valid email address
    company: Optional[str] = None      # Optional: Company name
    phone: Optional[str] = None        # Optional: Phone number
    linkedin_url: Optional[str] = None # Optional: LinkedIn profile URL
    job_title: Optional[str] = None    # Optional: Job title
    location: Optional[str] = None     # Optional: Location/city
    notes: Optional[str] = None        # Optional: Additional notes
    tags: Optional[List[str]] = None   # Optional: Array of tags
    custom_fields: Optional[Dict[str, Any]] = None  # Optional: Custom fields

Lead  # Complete lead object with generated ID and timestamps

try:
    # Create a new lead
    lead_request = buena_sdk.CreateLeadRequest(
        first_name="John",
        last_name="Doe",
        email="john.doe@example.com",
        company="Example Corp",
        phone="+1-555-123-4567",
        linkedin_url="https://linkedin.com/in/johndoe",
        job_title="Software Engineer",
        location="San Francisco, CA",
        notes="Met at tech conference 2024",
        tags=["conference", "engineer", "potential"],
        custom_fields={
            "lead_source": "conference",
            "priority": "high",
            "budget": 50000
        }
    )

    new_lead = api_instance.create_lead(lead_request)
    print(f"Created lead with ID: {new_lead.id}")
    print(f"Name: {new_lead.first_name} {new_lead.last_name}")
    print(f"Created at: {new_lead.created_at}")

except ApiException as e:
    print(f"Exception when creating lead: {e}")
    if e.status == 400:
        print("Check required fields: first_name, last_name, email")
    elif e.status == 409:
        print("Lead with this email already exists")

def get_lead(lead_id: str) -> Lead

Lead  # Complete lead object

try:
    lead = api_instance.get_lead("lead_123456789")
    print(f"Lead: {lead.first_name} {lead.last_name}")
    print(f"Email: {lead.email}")
    print(f"Company: {lead.company}")
    print(f"Status: {lead.status}")
    print(f"Created: {lead.created_at.strftime('%Y-%m-%d %H:%M:%S')}")

    # Access custom fields
    if lead.custom_fields:
        print("Custom fields:")
        for key, value in lead.custom_fields.items():
            print(f"  {key}: {value}")

except ApiException as e:
    print(f"Exception when getting lead: {e}")
    if e.status == 404:
        print("Lead not found")

def update_lead(
    lead_id: str,
    update_lead_request: UpdateLeadRequest
) -> Lead

class UpdateLeadRequest:
    first_name: Optional[str] = None            # Optional: Update first name
    last_name: Optional[str] = None             # Optional: Update last name
    email: Optional[str] = None                 # Optional: Update email (must be unique)
    company: Optional[str] = None               # Optional: Update company
    phone: Optional[str] = None                 # Optional: Update phone
    linkedin_url: Optional[str] = None          # Optional: Update LinkedIn URL
    job_title: Optional[str] = None             # Optional: Update job title
    location: Optional[str] = None              # Optional: Update location
    notes: Optional[str] = None                 # Optional: Update notes
    tags: Optional[List[str]] = None            # Optional: Replace tags array
    custom_fields: Optional[Dict[str, Any]] = None  # Optional: Update custom fields
    status: Optional[str] = None                # Optional: Update status

Lead  # Updated lead object

try:
    # Update lead information
    update_request = buena_sdk.UpdateLeadRequest(
        company="New Company Inc",
        job_title="Senior Software Engineer",
        status="contacted",
        tags=["contacted", "promoted", "hot-lead"],
        notes="Promoted to senior role, very interested in our product",
        custom_fields={
            "priority": "very_high",
            "last_contacted": "2024-06-29",
            "budget": 75000
        }
    )

    updated_lead = api_instance.update_lead("lead_123456789", update_request)
    print(f"Updated lead: {updated_lead.first_name} {updated_lead.last_name}")
    print(f"New company: {updated_lead.company}")
    print(f"New status: {updated_lead.status}")

except ApiException as e:
    print(f"Exception when updating lead: {e}")
    if e.status == 404:
        print("Lead not found")
    elif e.status == 409:
        print("Email already exists (if updating email)")

def delete_lead(lead_id: str) -> Dict[str, Any]

{
    "success": bool,         # Whether deletion was successful
    "message": str,          # Success/error message
    "deleted_id": str        # ID of deleted lead
}

try:
    result = api_instance.delete_lead("lead_123456789")
    if result.get("success"):
        print(f"Successfully deleted lead: {result.get('deleted_id')}")
        print(f"Message: {result.get('message')}")
    else:
        print(f"Failed to delete lead: {result.get('message')}")

except ApiException as e:
    print(f"Exception when deleting lead: {e}")
    if e.status == 404:
        print("Lead not found")

class ListApiKeysResponse:
    data: List[ApiKey]         # List of API key objects
    total: int                 # Total number of API keys

class ApiKey:
    id: str                    # Unique key identifier
    name: str                  # Descriptive name
    key_preview: str           # First 8 chars + "..."
    permissions: List[str]     # Array of permissions
    last_used: Optional[datetime]  # Last usage timestamp
    is_active: bool            # Whether key is active
    created_at: datetime       # Creation timestamp
    expires_at: Optional[datetime]  # Optional expiration date

try:
    api_keys = api_instance.list_api_keys()
    print(f"You have {api_keys.total} API keys:")

    for key in api_keys.data:
        status = "Active" if key.is_active else "Inactive"
        print(f"{key.name}: {key.key_preview} ({status})")
        print(f"  Permissions: {', '.join(key.permissions)}")
        if key.last_used:
            print(f"  Last used: {key.last_used.strftime('%Y-%m-%d %H:%M:%S')}")
        if key.expires_at:
            print(f"  Expires: {key.expires_at.strftime('%Y-%m-%d')}")
        print()

except ApiException as e:
    print(f"Exception when listing API keys: {e}")

def create_api_key(
    create_api_key_request: CreateApiKeyRequest
) -> ApiKey

class CreateApiKeyRequest:
    name: str                          # Required: Descriptive name
    permissions: List[str]             # Required: Array of permissions
    expires_at: Optional[datetime] = None  # Optional: Expiration date
    description: Optional[str] = None  # Optional: Key description

# Available permissions:
PERMISSIONS = [
    "leads:read",        # Read leads
    "leads:write",       # Create/update leads
    "leads:delete",      # Delete leads
    "api_keys:read",     # Read API keys
    "api_keys:write",    # Create API keys
    "webhooks:read",     # Read webhooks
    "webhooks:write"     # Create/update webhooks
]

class ApiKey:
    id: str                    # Unique key identifier
    name: str                  # Descriptive name
    key: str                   # Full API key (ONLY returned on creation!)
    key_preview: str           # First 8 chars + "..."
    permissions: List[str]     # Array of permissions
    is_active: bool            # Always True for new keys
    created_at: datetime       # Creation timestamp
    expires_at: Optional[datetime]  # Optional expiration date

from datetime import datetime, timedelta

try:
    # Create API key that expires in 1 year
    expiry_date = datetime.now() + timedelta(days=365)

    key_request = buena_sdk.CreateApiKeyRequest(
        name="Integration API Key",
        permissions=["leads:read", "leads:write"],
        description="For CRM integration",
        expires_at=expiry_date
    )

    new_api_key = api_instance.create_api_key(key_request)
    print(f"Created API key: {new_api_key.name}")
    print(f"Key: {new_api_key.key}")  # SAVE THIS! Won't be shown again
    print(f"Preview: {new_api_key.key_preview}")
    print(f"Permissions: {', '.join(new_api_key.permissions)}")
    print(f"Expires: {new_api_key.expires_at.strftime('%Y-%m-%d') if new_api_key.expires_at else 'Never'}")

except ApiException as e:
    print(f"Exception when creating API key: {e}")
    if e.status == 400:
        print("Check required fields: name, permissions")
    elif e.status == 422:
        print("Invalid permissions specified")

class HealthCheck200Response:
    status: str                # 'healthy', 'degraded', or 'down'
    timestamp: datetime        # Current server timestamp
    version: str               # API version
    services: Dict[str, str]   # Service health status
    response_time: int         # Response time in milliseconds

# Services dictionary structure:
{
    "database": "healthy|degraded|down",
    "redis": "healthy|degraded|down",
    "api": "healthy|degraded|down"
}

try:
    health = api_instance.health_check()
    print(f"API Status: {health.status}")
    print(f"Version: {health.version}")
    print(f"Response Time: {health.response_time}ms")
    print(f"Timestamp: {health.timestamp.strftime('%Y-%m-%d %H:%M:%S')}")

    if health.status == 'healthy':
        print("✅ All systems operational")
    else:
        print("⚠️  Some services may be experiencing issues")
        print("Service status:")
        for service, status in health.services.items():
            emoji = "✅" if status == "healthy" else "⚠️" if status == "degraded" else "❌"
            print(f"  {emoji} {service}: {status}")

except ApiException as e:
    print(f"Exception when checking health: {e}")
    if e.status == 503:
        print("Service temporarily unavailable")

import asyncio
import buena_sdk
from buena_sdk.api.default_api import DefaultApi
from buena_sdk.rest import ApiException

async def async_example():
    configuration = buena_sdk.Configuration(
        host="https://api.buena.ai/api/v2"
    )
    configuration.api_key['ApiKeyAuth'] = 'your-api-key'

    async with buena_sdk.ApiClient(configuration) as api_client:
        api_instance = DefaultApi(api_client)

        try:
            # List leads asynchronously
            leads = await api_instance.list_leads_async()
            print(f"Found {leads.total} leads")

            # Create lead asynchronously
            lead_request = buena_sdk.CreateLeadRequest(
                first_name="Jane",
                last_name="Smith",
                email="jane@example.com"
            )
            new_lead = await api_instance.create_lead_async(lead_request)
            print(f"Created lead: {new_lead.id}")

        except ApiException as e:
            print(f"Exception: {e}")

# Run async function
asyncio.run(async_example())

from buena_sdk.rest import ApiException
import logging

def handle_api_errors(func):
    """Decorator for handling API errors"""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            logging.error(f"API Exception in {func.__name__}: {e}")

            # Handle specific error codes
            error_messages = {
                401: "Invalid API key - check your authentication",
                403: "Insufficient permissions for this operation",
                404: "Resource not found",
                409: "Conflict - resource already exists",
                422: "Validation error - check your input data",
                429: "Rate limit exceeded - please retry later",
                500: "Server error - contact support if this persists",
                503: "Service temporarily unavailable"
            }

            message = error_messages.get(e.status, f"HTTP error {e.status}")
            print(f"Error {e.status}: {message}")

            # Log response body for debugging
            if hasattr(e, 'body') and e.body:
                logging.debug(f"Response body: {e.body}")

            raise  # Re-raise for caller to handle
        except Exception as e:
            logging.error(f"Unexpected error in {func.__name__}: {e}")
            raise
    return wrapper

# Usage example
@handle_api_errors
def create_lead_safe(api_instance, lead_data):
    return api_instance.create_lead(lead_data)

# Retry with exponential backoff
import time
import random

def api_call_with_retry(func, *args, max_retries=3, **kwargs):
    """Execute API call with exponential backoff retry"""
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            if e.status == 429 and attempt < max_retries - 1:
                # Rate limited - wait with exponential backoff
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited. Waiting {wait_time:.2f}s before retry {attempt + 1}...")
                time.sleep(wait_time)
            else:
                raise

# Usage
try:
    leads = api_call_with_retry(api_instance.list_leads, limit=50)
except ApiException as e:
    print(f"Failed after retries: {e}")

def paginate_leads(api_instance, page_size=50, **filters):
    """Generator function to paginate through all leads"""
    offset = 0

    while True:
        try:
            response = api_instance.list_leads(
                limit=page_size,
                offset=offset,
                **filters
            )

            # Yield each lead
            for lead in response.data:
                yield lead

            # Check if we've reached the end
            if len(response.data) < page_size:
                break

            offset += page_size

        except ApiException as e:
            print(f"Error during pagination: {e}")
            break

# Usage
for lead in paginate_leads(api_instance, company="Example Corp"):
    print(f"Processing lead: {lead.first_name} {lead.last_name}")

import buena_sdk
from urllib3.util.retry import Retry
from urllib3 import PoolManager

# Advanced configuration
configuration = buena_sdk.Configuration(
    host="https://api.buena.ai/api/v2",
    api_key={'ApiKeyAuth': 'your-api-key'},

    # SSL settings
    verify_ssl=True,
    ssl_ca_cert=None,
    cert_file=None,
    key_file=None,

    # Connection pool settings
    connection_pool_maxsize=10,

    # Proxy settings
    proxy=None,  # or "http://proxy.example.com:8080"
    proxy_headers=None,

    # Timeout settings
    timeout=30,  # Request timeout in seconds

    # User agent
    user_agent="BuenaSDK-Python/1.0.0"
)

# Custom retry strategy
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504],
)

# Use custom configuration
with buena_sdk.ApiClient(configuration) as api_client:
    # Configure custom retry strategy
    api_client.rest_client.pool_manager = PoolManager(
        num_pools=10,
        retries=retry_strategy
    )

    api_instance = DefaultApi(api_client)