Use FFmpeg API in Python

Use the requests library to submit a video processing job. This example transcodes a video to 720p:

import os
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv('FFMPEG_API_KEY')
BASE_URL = 'https://api.ffmpeg-micro.com/v1'

def submit_transcode_job(input_url, output_format='mp4'):
    """Submit a video transcoding job to FFmpeg Micro API."""
    headers = {
        'Authorization': f'Bearer {API_KEY}',
        'Content-Type': 'application/json'
    }

    payload = {
        'inputs': [{'url': input_url}],
        'outputFormat': output_format,
        'options': [
            {'option': '-s', 'argument': '1280x720'},
            {'option': '-c:v', 'argument': 'libx264'},
            {'option': '-preset', 'argument': 'fast'},
            {'option': '-crf', 'argument': '23'}
        ]
    }

    response = requests.post(
        f'{BASE_URL}/transcodes',
        headers=headers,
        json=payload
    )
    response.raise_for_status()

    return response.json()

# Example usage
job = submit_transcode_job('https://example.com/video.mp4')
print(f"Job submitted: {job['id']}")
print(f"Status: {job['status']}")

The API returns a job ID and initial status. Processing happens asynchronously.

Receive and process webhook callbacks in Flask:

from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)

@app.route('/webhooks/ffmpeg', methods=['POST'])
def handle_ffmpeg_webhook():
    """Handle completion webhook from FFmpeg Micro."""
    payload = request.json

    # Verify webhook signature (recommended)
    signature = request.headers.get('X-FFmpeg-Signature')
    if not verify_signature(request.data, signature):
        return jsonify({'error': 'Invalid signature'}), 401

    job_id = payload['id']
    status = payload['status']

    if status == 'completed':
        output_url = payload.get('outputUrl')
        print(f"Job {job_id} completed: {output_url}")

        # Process the result (upload to S3, update database, etc.)
        process_completed_video(job_id, output_url)

    elif status == 'failed':
        error = payload.get('error', 'Unknown error')
        print(f"Job {job_id} failed: {error}")
        handle_failure(job_id, error)

    return jsonify({'received': True}), 200

def verify_signature(payload_bytes, signature):
    """Verify webhook signature using your webhook secret."""
    webhook_secret = os.getenv('FFMPEG_WEBHOOK_SECRET')
    expected = hmac.new(
        webhook_secret.encode(),
        payload_bytes,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

def process_completed_video(job_id, output_url):
    """Your custom logic for completed videos."""
    pass

def handle_failure(job_id, error):
    """Your custom logic for failed jobs."""
    pass

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

Or use FastAPI for async webhook processing:

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
import hmac
import hashlib

app = FastAPI()

class WebhookPayload(BaseModel):
    id: str
    status: str
    outputUrl: str | None = None
    error: str | None = None

@app.post('/webhooks/ffmpeg')
async def handle_ffmpeg_webhook(
    request: Request,
    payload: WebhookPayload
):
    """Handle FFmpeg job completion webhook."""

    # Verify signature
    signature = request.headers.get('x-ffmpeg-signature')
    body = await request.body()

    if not verify_signature(body, signature):
        raise HTTPException(status_code=401, detail='Invalid signature')

    if payload.status == 'completed':
        print(f"Job {payload.id} completed: {payload.outputUrl}")
        # Process result asynchronously
        await process_completed_video(payload.id, payload.outputUrl)

    elif payload.status == 'failed':
        print(f"Job {payload.id} failed: {payload.error}")
        await handle_failure(payload.id, payload.error)

    return {'received': True}

def verify_signature(payload_bytes: bytes, signature: str) -> bool:
    """Verify webhook signature."""
    webhook_secret = os.getenv('FFMPEG_WEBHOOK_SECRET')
    expected = hmac.new(
        webhook_secret.encode(),
        payload_bytes,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature or '')

async def process_completed_video(job_id: str, output_url: str):
    """Your async processing logic."""
    pass

async def handle_failure(job_id: str, error: str):
    """Your async error handling logic."""
    pass

Process multiple videos in parallel:

import concurrent.futures

def process_video_batch(video_urls, max_workers=5):
    """Submit multiple transcoding jobs in parallel."""

    def submit_job(url):
        try:
            job = submit_transcode_job(url)
            return {'url': url, 'job_id': job['id'], 'status': 'submitted'}
        except Exception as e:
            return {'url': url, 'error': str(e), 'status': 'failed'}

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(submit_job, video_urls))

    return results

# Example: process a batch of videos
video_urls = [
    'https://example.com/video1.mp4',
    'https://example.com/video2.mp4',
    'https://example.com/video3.mp4',
]

results = process_video_batch(video_urls)

for result in results:
    if result['status'] == 'submitted':
        print(f"Submitted {result['url']}: Job ID {result['job_id']}")
    else:
        print(f"Failed {result['url']}: {result['error']}")

Use webhooks for batch jobs instead of polling each job individually. This scales better for large batches.

Handle common errors and edge cases:

import requests
from requests.exceptions import RequestException, HTTPError, Timeout

def submit_with_retry(input_url, max_retries=3):
    """Submit job with automatic retry on network errors."""

    for attempt in range(max_retries):
        try:
            job = submit_transcode_job(input_url)
            return job

        except Timeout:
            print(f"Attempt {attempt + 1}: Request timeout")
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff

        except HTTPError as e:
            if e.response.status_code == 401:
                raise Exception("Invalid API key")
            elif e.response.status_code == 402:
                raise Exception("Insufficient credits")
            elif e.response.status_code == 422:
                error_detail = e.response.json()
                raise Exception(f"Invalid request: {error_detail}")
            elif e.response.status_code >= 500:
                print(f"Attempt {attempt + 1}: Server error")
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
            else:
                raise

        except RequestException as e:
            print(f"Attempt {attempt + 1}: Network error: {e}")
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

# Example with error handling
try:
    job = submit_with_retry('https://example.com/video.mp4')
    print(f"Job submitted successfully: {job['id']}")
except Exception as e:
    print(f"Failed to submit job: {e}")