Python SDK

pip install bluma

# Or with Poetry
poetry add bluma

from bluma import Bluma

bluma = Bluma(api_key=os.getenv('BLUMA_API_KEY'))

# Generate a video
video = bluma.videos.create(
    template_id='rick-morty-explainer',
    context={
        'prompt': 'Create a funny dialogue between a programmer and their computer'
    }
)

print(f'Video created: {video.id}')

# Wait for completion
completed = bluma.videos.wait_for(video.id)
print(f'Video ready: {completed.url}')

from bluma import Bluma

bluma = Bluma(
    api_key=os.getenv('BLUMA_API_KEY'),
    base_url='https://api.getbluma.com/api/v1',  # Optional, default shown
    timeout=30,  # seconds
    max_retries=3,
    retry_delay=1.0  # seconds
)

import os
from bluma import Bluma

env = os.getenv('ENVIRONMENT', 'development')

config = {
    'development': {
        'api_key': os.getenv('BLUMA_TEST_KEY'),
        'timeout': 60
    },
    'production': {
        'api_key': os.getenv('BLUMA_LIVE_KEY'),
        'timeout': 30
    }
}

bluma = Bluma(**config[env])

# Using a template directly
video = bluma.videos.create(
    template_id='rick-morty-explainer',
    context={'prompt': 'Create a funny cat dialogue'}
)

# Using a variant preset (saved configuration)
video = bluma.videos.create(
    variant_id='var_xyz789',  # Uses pre-configured settings
    context={'prompt': 'Create a funny cat dialogue'}
)

# With all optional parameters
video = bluma.videos.create(
    template_id='ugc-text-overlay',
    context={
        'prompt': 'Showcase my new product',
        'brand_assets': {'logo': 'https://cdn.example.com/logo.png'},
        'system_prompt': 'You are an enthusiastic marketer'
    },
    options={
        'resolution': '4k',       # '720p', '1080p', or '4k'
        'watermark': False        # Add watermark (test keys only)
    },
    webhook_url='https://myapp.com/webhooks/bluma'
)

print(video.id)              # 'batch_abc123xyz'
print(video.status)          # 'queued'
print(video.template_id)     # 'ugc-text-overlay'
print(video.variant_id)      # 'var_xyz789' or None
print(video.credits_charged) # 6

from typing import Dict, Any, Optional
from bluma.types import Video, VideoStatus

def create_video(
    context: Dict[str, Any],
    template_id: Optional[str] = None,
    variant_id: Optional[str] = None,
    options: Optional[Dict[str, Any]] = None,
    webhook_url: Optional[str] = None
) -> Video:
    """Either template_id or variant_id must be provided"""
    return bluma.videos.create(
        context=context,
        template_id=template_id,
        variant_id=variant_id,
        options=options,
        webhook_url=webhook_url
    )

video = bluma.videos.get('batch_abc123xyz')

print(video.status)    # 'completed'
print(video.url)       # 'https://cdn.getbluma.com/videos/...'
print(video.duration)  # 45 (seconds)

# Polls until video is completed or failed
video = bluma.videos.wait_for(
    'batch_abc123xyz',
    poll_interval=5,  # seconds (default: 5)
    timeout=300       # seconds (default: 600)
)

if video.status == VideoStatus.COMPLETED:
    print(f'Video ready: {video.url}')
else:
    print(f'Video failed: {video.error}')

def on_progress(progress: int):
    print(f'Progress: {progress}%')

video = bluma.videos.wait_for(
    'batch_abc123xyz',
    on_progress=on_progress
)

download = bluma.videos.download('batch_abc123xyz')

print(download.download_url)  # Presigned URL
print(download.expires_at)    # Expiration datetime

# Download to file
import requests

response = requests.get(download.download_url)
with open('my-video.mp4', 'wb') as f:
    f.write(response.content)

templates = bluma.templates.list()

for template in templates:
    print(f'{template.id}: {template.name}')
    print(f'Base cost: {template.base_cost} credits')

template = bluma.templates.get('rick-morty-explainer')

print(template.name)           # 'Rick & Morty Meme Explainer'
print(template.description)    # 'Create funny dialogue videos...'
print(template.context_schema) # JSON schema for context

balance = bluma.credits.get_balance()

print(balance.credits)           # 88
print(balance.tier)              # 'pro'
print(balance.monthly_allowance) # 500
print(balance.reset_date)        # datetime object

history = bluma.credits.get_history(limit=50, offset=0)

for txn in history.transactions:
    print(txn.type)         # 'deduction' | 'purchase'
    print(txn.amount)       # -6 or +100
    print(txn.description)  # 'Video generation: batch_xyz789'
    print(txn.created_at)   # datetime object

print(f'Total transactions: {history.total}')

from bluma import verify_webhook
from flask import Flask, request
import os

app = Flask(__name__)

@app.route('/webhooks/bluma', methods=['POST'])
def webhook():
    signature = request.headers.get('X-Bluma-Signature')
    payload = request.get_data()

    try:
        # Verify and parse webhook
        event = verify_webhook(
            payload,
            signature,
            os.getenv('BLUMA_WEBHOOK_SECRET')
        )

        # Handle event
        if event.type == 'video.completed':
            print(f'Video ready: {event.data.url}')

        elif event.type == 'video.failed':
            print(f'Video failed: {event.data.error}')

        elif event.type == 'credits.low':
            print(f'Credits low: {event.data.remaining}')

        return '', 200

    except Exception as error:
        print('Invalid webhook signature')
        return 'Unauthorized', 401

if __name__ == '__main__':
    app.run(port=3000)

webhook = bluma.webhooks.create(
    url='https://myapp.com/webhooks/bluma',
    events=['video.completed', 'video.failed']
)

print(webhook.id)     # 'webhook_abc123'
print(webhook.secret) # 'whsec_...' (save this!)

webhooks = bluma.webhooks.list()

for webhook in webhooks:
    print(f'{webhook.id}: {webhook.url}')
    print(f'Active: {webhook.is_active}')

bluma.webhooks.delete('webhook_abc123')

variant = bluma.variants.create(
    template_id='rick-morty-explainer',
    name='Funny Tone Preset',
    settings={
        'systemPrompt': 'Use a funny, lighthearted tone',
        'captionPrompt': 'Create engaging captions with emojis',
        'compositionProps': {
            'voiceId': 'female-casual',
            'primaryColor': '#FF69B4'
        }
    }
)

print(variant.id)  # 'var_xyz789'

variants = bluma.variants.list('rick-morty-explainer')

for variant in variants:
    print(f'{variant.name}: {variant.is_active}')

variant = bluma.variants.get('rick-morty-explainer', 'var_xyz789')

print(variant.payload)  # Contains saved settings

updated = bluma.variants.update(
    template_id='rick-morty-explainer',
    variant_id='var_xyz789',
    settings={'systemPrompt': 'Updated tone instructions'}
)

bluma.variants.delete('rick-morty-explainer', 'var_xyz789')

collection = bluma.collections.create(
    name='Product Photos',
    description='High-quality product photography'
)

print(collection.id)          # 'coll_abc123'
print(collection.asset_count) # 0

collections = bluma.collections.list()

for collection in collections:
    print(f'{collection.name}: {collection.asset_count} assets')

collection = bluma.collections.get('coll_abc123')

print(collection.name)
print(collection.description)
print(collection.asset_count)

updated = bluma.collections.rename('coll_abc123', 'Updated Collection Name')

bluma.collections.add_assets(
    collection_id='coll_abc123',
    asset_ids=['asset_1', 'asset_2', 'asset_3']
)

bluma.collections.remove_asset('coll_abc123', 'asset_1')

assets = bluma.collections.list_assets('coll_abc123')

for asset in assets:
    print(f'{asset.name}: {asset.cdn_url}')

bluma.collections.delete('coll_abc123')

import requests

# 1. Get presigned upload URL
upload_response = bluma.assets.upload(
    file_name='product.jpg',
    file_type='image/jpeg',
    collection_ids=['coll_abc123']  # Optional
)

print(upload_response.asset_id)  # 'asset_abc123'
print(upload_response.cdn_url)   # Final CDN URL

# 2. Upload file to presigned URL
with open('product.jpg', 'rb') as f:
    requests.put(upload_response.upload_url, data=f)

print('Upload complete!')

asset = bluma.assets.get('asset_abc123')

print(asset.name)
print(asset.cdn_url)
print(asset.file_type)
print(asset.file_size_bytes)

# List all assets
assets = bluma.assets.list()

# Filter by type
images = bluma.assets.list(file_type='image')

# Filter by collection
collection_assets = bluma.assets.list(collection_id='coll_abc123')

# Include deleted assets
all_assets = bluma.assets.list(include_deleted=True)

for asset in assets:
    print(f'{asset.name} ({asset.file_type})')

# Get random asset from collection
random_asset = bluma.assets.get_random(
    file_type='image',
    collection_id='coll_abc123',
    used_asset_ids=['asset_1', 'asset_2']  # Exclude these
)

print(random_asset.cdn_url)

updated = bluma.assets.rename('asset_abc123', 'New Asset Name')

bluma.assets.delete('asset_abc123')

recovered = bluma.assets.recover('asset_abc123')
print(f'Asset recovered: {recovered.name}')

from bluma.errors import (
    BlumaError,
    ValidationError,
    AuthenticationError,
    InsufficientCreditsError,
    RateLimitError,
    NotFoundError,
    APIError
)

try:
    video = bluma.videos.create(
        template_id='invalid-template',
        context={'prompt': 'Test'}
    )
except ValidationError as error:
    print(f'Invalid input: {error.message}')
    print(f'Field: {error.field}')
except AuthenticationError:
    print('Invalid API key')
except InsufficientCreditsError as error:
    print('Out of credits!')
    print(f'Required: {error.credits_required}')
    print(f'Available: {error.credits_available}')
except RateLimitError as error:
    print(f'Rate limited. Retry after {error.retry_after} seconds')
except NotFoundError:
    print('Resource not found')
except APIError as error:
    print(f'API error: {error.message}')
    print(f'Status: {error.status}')
    print(f'Type: {error.type}')
except Exception as error:
    print(f'Network error: {error}')

# Retries are enabled by default
bluma = Bluma(
    api_key=os.getenv('BLUMA_API_KEY'),
    max_retries=3,         # Number of retries (default: 3)
    retry_delay=1.0,       # Initial delay in seconds (default: 1.0)
    retry_multiplier=2.0   # Exponential backoff multiplier (default: 2.0)
)

# Retry sequence: 1s, 2s, 4s
# Only retries on 5xx errors and network failures

from bluma import Bluma

def should_retry(error, attempt_number):
    """Custom retry logic"""
    if isinstance(error, RateLimitError):
        return attempt_number < 5  # Retry rate limits up to 5 times
    if isinstance(error, ValidationError):
        return False  # Never retry validation errors
    return attempt_number < 3  # Default: retry 3 times

def calculate_retry_delay(attempt_number):
    """Custom backoff"""
    return min(1.0 * (2 ** attempt_number), 10.0)

bluma = Bluma(
    api_key=os.getenv('BLUMA_API_KEY'),
    should_retry=should_retry,
    calculate_retry_delay=calculate_retry_delay
)

from concurrent.futures import ThreadPoolExecutor

templates = ['rick-morty-explainer', 'ugc-text-overlay', 'cat-explainer-v2']

# Generate videos in parallel
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [
        executor.submit(
            bluma.videos.create,
            template_id=template_id,
            context={'prompt': f'Create a {template_id} video'}
        )
        for template_id in templates
    ]

    videos = [f.result() for f in futures]

print(f'Created {len(videos)} videos')

from bluma import Bluma

# Automatically closes HTTP session
with Bluma(api_key=os.getenv('BLUMA_API_KEY')) as client:
    video = client.videos.create(
        template_id='rick-morty-explainer',
        context={'prompt': 'Test'}
    )
    print(video.id)

# Session is closed here

bluma = Bluma(
    api_key=os.getenv('BLUMA_API_KEY'),
    default_headers={
        'User-Agent': 'MyApp/1.0',
        'X-Custom-Header': 'value'
    }
)

from bluma.types import (
    Video,
    VideoStatus,
    Template,
    CreditBalance,
    Transaction,
    Webhook,
    WebhookEvent,
    TemplateVariant,
    Collection,
    Asset,
    AssetUploadResponse
)

def process_video(video: Video) -> None:
    if video.status == VideoStatus.COMPLETED:
        print(video.url)

def upload_asset(file_path: str) -> Asset:
    upload_resp: AssetUploadResponse = bluma.assets.upload(
        file_name=file_path,
        file_type='image/jpeg'
    )
    return bluma.assets.get(upload_resp.asset_id)

from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class Video:
    id: str
    status: VideoStatus
    template_id: str
    url: Optional[str] = None
    thumbnail_url: Optional[str] = None
    duration: Optional[int] = None
    size_bytes: Optional[int] = None
    credits_charged: int = 0
    created_at: datetime
    completed_at: Optional[datetime] = None
    error: Optional[dict] = None

import os
from bluma import Bluma
from bluma.errors import InsufficientCreditsError

def generate_and_download():
    bluma = Bluma(api_key=os.getenv('BLUMA_API_KEY'))

    try:
        # 1. Create video
        print('Creating video...')
        video = bluma.videos.create(
            template_id='rick-morty-explainer',
            context={'prompt': 'Create a funny programmer dialogue'}
        )

        print(f'Video ID: {video.id}')
        print(f'Credits charged: {video.credits_charged}')

        # 2. Wait for completion
        print('Waiting for video to complete...')

        def on_progress(progress):
            print(f'Progress: {progress}%')

        completed = bluma.videos.wait_for(
            video.id,
            on_progress=on_progress
        )

        print('Video completed!')

        # 3. Download video
        download = bluma.videos.download(completed.id)

        # 4. Save to file
        import requests
        response = requests.get(download.download_url)

        with open('my-video.mp4', 'wb') as f:
            f.write(response.content)

        print('Video saved to my-video.mp4')

    except InsufficientCreditsError:
        print('Out of credits! Please purchase more.')
    except Exception as error:
        print(f'Error: {error}')

if __name__ == '__main__':
    generate_and_download()

import os
import requests
from bluma import Bluma
from bluma.errors import BlumaError

def setup_asset_driven_videos():
    """Complete workflow: Upload assets and save a configuration preset"""
    bluma = Bluma(api_key=os.getenv('BLUMA_API_KEY'))

    try:
        # 1. Create an asset collection
        print('Creating asset collection...')
        collection = bluma.collections.create(
            name='Product Images',
            description='Product photography for videos'
        )

        # 2. Upload assets to collection
        print('Uploading assets...')
        assets = []
        for image_file in ['product1.jpg', 'product2.jpg', 'product3.jpg']:
            # Get presigned upload URL
            upload_resp = bluma.assets.upload(
                file_name=image_file,
                file_type='image/jpeg',
                collection_ids=[collection.id]
            )

            # Upload file
            with open(image_file, 'rb') as f:
                requests.put(upload_resp.upload_url, data=f)

            assets.append(upload_resp.asset_id)
            print(f'Uploaded {image_file} -> {upload_resp.cdn_url}')

        # 3. Create a variant preset for easy reuse
        print('Creating variant preset...')
        variant = bluma.variants.create(
            template_id='rick-morty-explainer',
            name='Product Marketing Preset',
            settings={
                'systemPrompt': 'You are an enthusiastic product marketer',
                'compositionProps': {
                    'voiceId': 'energetic-male',
                    'assetCollectionId': collection.id
                }
            }
        )

        print(f'✅ Setup complete!')
        print(f'Variant Preset: {variant.id}')
        print(f'Collection: {collection.id} ({len(assets)} assets)')

        # 4. Generate a test video using the variant preset
        print('Generating test video with variant...')
        video = bluma.videos.create(
            variant_id=variant.id,  # Use the saved preset
            context={
                'prompt': 'Create funny dialogue promoting our product'
            }
        )

        print(f'Test video queued: {video.id}')

        # 5. Wait for completion
        completed = bluma.videos.wait_for(video.id)
        print(f'Test video ready: {completed.url}')

        return {
            'variant_id': variant.id,
            'collection_id': collection.id,
            'test_video_url': completed.url
        }

    except BlumaError as error:
        print(f'Error: {error}')
        return None

if __name__ == '__main__':
    result = setup_asset_driven_videos()
    if result:
        print(f'Setup complete! Use the preset for future videos.')

from flask import Flask, request, jsonify
from bluma import Bluma
from bluma.errors import BlumaError, NotFoundError

app = Flask(__name__)
bluma = Bluma(api_key=os.getenv('BLUMA_API_KEY'))

@app.route('/api/videos', methods=['POST'])
def create_video():
    try:
        video = bluma.videos.create(
            template_id=request.json['template_id'],
            context=request.json['context']
        )
        return jsonify(video.__dict__), 200
    except BlumaError as error:
        return jsonify({'error': str(error)}), error.status
    except Exception:
        return jsonify({'error': 'Internal server error'}), 500

@app.route('/api/videos/<video_id>', methods=['GET'])
def get_video(video_id):
    try:
        video = bluma.videos.get(video_id)
        return jsonify(video.__dict__), 200
    except NotFoundError:
        return jsonify({'error': 'Video not found'}), 404
    except Exception:
        return jsonify({'error': 'Internal server error'}), 500

if __name__ == '__main__':
    app.run(port=3000)

# views.py
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from bluma import Bluma
from bluma.errors import BlumaError, NotFoundError
import os

bluma = Bluma(api_key=os.getenv('BLUMA_API_KEY'))

@require_http_methods(["POST"])
def create_video(request):
    try:
        import json
        data = json.loads(request.body)

        video = bluma.videos.create(
            template_id=data['template_id'],
            context=data['context']
        )

        return JsonResponse({
            'id': video.id,
            'status': video.status,
            'credits_charged': video.credits_charged
        })

    except BlumaError as error:
        return JsonResponse(
            {'error': str(error)},
            status=error.status
        )

@require_http_methods(["GET"])
def get_video(request, video_id):
    try:
        video = bluma.videos.get(video_id)

        return JsonResponse({
            'id': video.id,
            'status': video.status,
            'url': video.url
        })

    except NotFoundError:
        return JsonResponse(
            {'error': 'Video not found'},
            status=404
        )

from celery import Celery
from bluma import Bluma
import os

app = Celery('tasks', broker='redis://localhost:6379/0')
bluma = Bluma(api_key=os.getenv('BLUMA_API_KEY'))

@app.task
def generate_video(template_id, context):
    """Background task for video generation"""
    try:
        # Create video
        video = bluma.videos.create(
            template_id=template_id,
            context=context
        )

        # Wait for completion
        completed = bluma.videos.wait_for(video.id)

        # Download to storage
        download = bluma.videos.download(completed.id)

        return {
            'success': True,
            'video_id': completed.id,
            'url': download.download_url
        }

    except Exception as error:
        return {
            'success': False,
            'error': str(error)
        }

# Usage
result = generate_video.delay('rick-morty-explainer', {'prompt': 'Test'})

from unittest.mock import Mock, patch
from bluma import Bluma
from bluma.types import Video, VideoStatus

def test_video_creation():
    # Mock the API client
    mock_bluma = Mock(spec=Bluma)

    mock_video = Video(
        id='batch_abc123',
        status=VideoStatus.COMPLETED,
        template_id='rick-morty-explainer',
        url='https://cdn.getbluma.com/videos/mock.mp4',
        credits_charged=5,
        created_at=datetime.now()
    )

    mock_bluma.videos.create.return_value = mock_video

    # Test your code
    video = mock_bluma.videos.create(
        template_id='rick-morty-explainer',
        context={'prompt': 'Test'}
    )

    assert video.id == 'batch_abc123'
    assert video.status == VideoStatus.COMPLETED

import pytest
from bluma import Bluma

@pytest.fixture
def bluma_client():
    return Bluma(api_key=os.getenv('BLUMA_TEST_KEY'))

def test_list_templates(bluma_client):
    templates = bluma_client.templates.list()
    assert len(templates) > 0
    assert all(hasattr(t, 'id') for t in templates)