Security & Privacy

import videodb

# Secure connection
conn = videodb.connect()  # Uses VIDEO_DB_API_KEY env var

# Desktop capture with client tokens (never expose API key)
cap = conn.create_capture_session(
    end_user_id="user_abc",
    callback_url="https://your-backend.com/webhooks"
)
token = conn.generate_client_token(expires_in=600)  # Short-lived

video = coll.upload(url="https://...")  # Stored
understanding = video.understand(
    analyzers=[{"type": "vlm", "name": "scene"}]
)
understanding.wait_until_complete()
scene = understanding.get_analyzer("scene")
index = video.index(name="scene", source=scene)  # Retrieval-ready
index.wait_until_complete()

# Indexing a continuous live output requires stored records.
understanding = rtstream.understand(
    segmentation={"type": "time", "window": "5s"},
    analyzers=[
        {
            "type": "vlm",
            "name": "scene",
            "sampling": {"frame_count": 1},
            "config": {"prompt": "Monitor for activity"},
        }
    ],
    store=True,
)
index = rtstream.index(
    name="live_scene",
    source=understanding.outputs["scene"],
    use_for=["semantic"],
)

# Delete video and all associated data
video.delete()

# Delete specific index
index.delete()

# Delete collection
coll.delete()

# Environment variable (recommended)
conn = videodb.connect()  # Uses VIDEO_DB_API_KEY

# Explicit key (server-side only)
conn = videodb.connect(api_key="your-key")

# Backend (holds API key)
token = conn.generate_client_token(expires_in=600)  # 10 minutes

# Client (uses token, never sees API key)
client = CaptureClient(client_token=token)

# Request explicit permission
await client.request_permission("microphone")
await client.request_permission("screen_capture")

# User must grant permission in OS dialog
channels = await client.list_channels()

@app.post("/webhooks/videodb")
async def handle_webhook(request):
    # Verify signature (implementation depends on your setup)
    signature = request.headers.get("X-VideoDB-Signature")
    if not verify_signature(request.body, signature):
        return {"error": "Invalid signature"}, 401

    event = await request.json()
    process_event(event)
    return {"status": "ok"}