Upload a Video

This guide provides and explains a Python 3 script that uploads a YouTube video using the YouTube Data API videos.insert method and resumable media uploads. The script uses the Google API Client Library for Python and google-auth-oauthlib, with retry and exponential backoff handling for transient HTTP 5xx and network errors. (Client libraries for other programming languages are also available.)

Requirements

  • Python: Python 3.7 or higher.
  • Client and authorization libraries: Install google-api-python-client, google-auth-oauthlib, and google-auth-httplib2:

    pip install --upgrade google-api-python-client google-auth-oauthlib google-auth-httplib2
    
  • OAuth 2.0 credentials: Set up a project and enable the YouTube Data API in the Google Cloud console, create an OAuth 2.0 Desktop app client ID on the Credentials page, and download the JSON credentials file as client_secrets.json in the same directory as upload_video.py:

    {
      "installed": {
        "client_id": "YOUR_CLIENT_ID.apps.googleusercontent.com",
        "project_id": "YOUR_PROJECT_ID",
        "auth_uri": "https://accounts.google.com/o/oauth2/auth",
        "token_uri": "https://oauth2.googleapis.com/token",
        "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
        "client_secret": "YOUR_CLIENT_SECRET",
        "redirect_uris": ["http://localhost"]
      }
    }
    

    On the first run, the script opens a browser window so you can authorize the https://www.googleapis.com/auth/youtube.upload scope, then caches the resulting access and refresh tokens in token.json for subsequent runs.

Sample request

This command runs upload_video.py to upload a video and set its title, description, tags, category, and privacy status:

python3 upload_video.py --file="/tmp/test_video_file.mp4" \
                        --title="Summer vacation in California" \
                        --description="Had fun surfing in Santa Cruz" \
                        --keywords="surfing,Santa Cruz" \
                        --category="22" \
                        --privacyStatus="private"

In this example, the script constructs and inserts the following video resource:

{
  "snippet": {
    "title": "Summer vacation in California",
    "description": "Had fun surfing in Santa Cruz",
    "tags": ["surfing", "Santa Cruz"],
    "categoryId": "22"
  },
  "status": {
    "privacyStatus": "private"
  }
}

Run the script

The following list defines the command-line arguments accepted by upload_video.py:

  • --file: Path to the video file that you are uploading (required).

    Example: --file="/home/path/to/file.mp4"
    
  • --title: Title of the video. Defaults to Test Title.

    Example: --title="Summer vacation in California"
    
  • --description: Description of the video. Defaults to Test Description.

    Example: --description="Had fun surfing in Santa Cruz"
    
  • --category: Numeric ID for the YouTube video category associated with the video. Defaults to 22 (People & Blogs).

    Example: --category="22"
    
  • --keywords: Comma-separated list of tags associated with the video. Defaults to an empty string.

    Example: --keywords="surfing"
    
  • --privacyStatus: Privacy status of the uploaded video. Valid values are public, private, and unlisted. Defaults to public; pass private or unlisted when uploading test videos.

    Example: --privacyStatus="private"
    

Sample code

The following code block shows the complete upload_video.py script:

#!/usr/bin/env python3
"""Uploads a video to YouTube using the YouTube Data API v3."""

import argparse
import http.client
import os
import random
import sys
import time
from typing import Any

from google.auth.transport import requests as google_requests
from google.oauth2 import credentials as oauth2_credentials
from google_auth_oauthlib import flow as oauth_flow
from googleapiclient import discovery
from googleapiclient import errors
from googleapiclient import http as media_http
import httplib2


# Explicitly tell the underlying HTTP transport library not to retry, since
# we are handling retry logic ourselves.
httplib2.RETRIES = 1

# Maximum number of times to retry before giving up.
MAX_RETRIES = 10

# Always retry when these exceptions are raised.
RETRIABLE_EXCEPTIONS = (
    httplib2.HttpLib2Error,
    OSError,
    http.client.NotConnected,
    http.client.IncompleteRead,
    http.client.ImproperConnectionState,
    http.client.CannotSendRequest,
    http.client.CannotSendHeader,
    http.client.ResponseNotReady,
    http.client.BadStatusLine,
)

# Always retry when a googleapiclient.errors.HttpError with one of these
# status codes is raised.
RETRIABLE_STATUS_CODES = (500, 502, 503, 504)

# The CLIENT_SECRETS_FILE variable specifies the name of a file that contains
# the OAuth 2.0 information for this application, including its client_id and
# client_secret. Acquire an OAuth 2.0 client ID and client secret from the
# Google Cloud console at https://console.cloud.google.com/.
# Ensure that you have enabled the YouTube Data API v3 for your project.
# For more information about using OAuth 2.0 to access the YouTube Data API,
# see https://developers.google.com/youtube/data/guides/authentication.
CLIENT_SECRETS_FILE = 'client_secrets.json'
TOKEN_FILE = 'token.json'

# This OAuth 2.0 access scope grants upload access to the authenticated
# user's YouTube channel, without granting other types of account access.
YOUTUBE_SCOPES = ['https://www.googleapis.com/auth/youtube.upload']
YOUTUBE_API_SERVICE_NAME = 'youtube'
YOUTUBE_API_VERSION = 'v3'

VALID_PRIVACY_STATUSES = ('public', 'private', 'unlisted')


def get_authenticated_service() -> Any:
  """Authorizes the user and returns an authenticated YouTube API client."""
  creds = None
  if os.path.exists(TOKEN_FILE):
    creds = oauth2_credentials.Credentials.from_authorized_user_file(
        TOKEN_FILE, YOUTUBE_SCOPES
    )

  if not creds or not creds.valid:
    if creds and creds.expired and creds.refresh_token:
      creds.refresh(google_requests.Request())
    else:
      if not os.path.exists(CLIENT_SECRETS_FILE):
        sys.exit(
            f'Missing {CLIENT_SECRETS_FILE}. Download OAuth 2.0 Desktop app '
            'credentials from https://console.cloud.google.com/apis/credentials '
            f'and save them to {os.path.abspath(CLIENT_SECRETS_FILE)}.'
        )
      app_flow = oauth_flow.InstalledAppFlow.from_client_secrets_file(
          CLIENT_SECRETS_FILE, YOUTUBE_SCOPES
      )
      creds = app_flow.run_local_server(port=0)
    with open(TOKEN_FILE, 'w', encoding='utf-8') as token_file:
      token_file.write(creds.to_json())

  return discovery.build(
      YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, credentials=creds
  )


def resumable_upload(insert_request: media_http.HttpRequest) -> None:
  """Executes a resumable upload with exponential backoff on transient errors.

  Args:
    insert_request: Configured YouTube videos.insert HttpRequest instance.
  """
  upload_response = None
  retry = 0
  while upload_response is None:
    error_message = None
    try:
      print('Uploading file...')
      _, upload_response = insert_request.next_chunk()
      if upload_response is not None:
        if 'id' in upload_response:
          video_id = upload_response['id']
          print(f"Video id '{video_id}' was successfully uploaded.")
        else:
          sys.exit(
              f'The upload failed with an unexpected response: '
              f'{upload_response}'
          )
    except errors.HttpError as err:
      if err.resp.status in RETRIABLE_STATUS_CODES:
        error_message = (
            f'A retriable HTTP error {err.resp.status} occurred:\n{err.content}'
        )
      else:
        raise
    except RETRIABLE_EXCEPTIONS as err:
      error_message = f'A retriable error occurred: {err}'

    if error_message is not None:
      print(error_message)
      retry += 1
      if retry > MAX_RETRIES:
        sys.exit('No longer attempting to retry.')

      max_sleep = 2**retry
      sleep_seconds = random.random() * max_sleep
      print(f'Sleeping {sleep_seconds:f} seconds and then retrying...')
      time.sleep(sleep_seconds)


def initialize_upload(youtube: Any, options: argparse.Namespace) -> None:
  """Builds and starts the YouTube videos.insert resumable upload request.

  Args:
    youtube: Authenticated YouTube Data API v3 resource client.
    options: Parsed command-line arguments describing the video metadata.
  """
  tags = options.keywords.split(',') if options.keywords else None

  request_body = {
      'snippet': {
          'title': options.title,
          'description': options.description,
          'tags': tags,
          'categoryId': options.category,
      },
      'status': {'privacyStatus': options.privacyStatus},
  }

  # Call the API's videos.insert method to create and upload the video.
  # Setting chunksize to -1 uploads the entire file in a single HTTP request
  # while still allowing failed uploads to resume where they left off. Use a
  # positive multiple of 256 KB (such as 1024 * 1024) on unstable connections
  # to transfer smaller chunks.
  insert_request = youtube.videos().insert(
      part=','.join(request_body.keys()),
      body=request_body,
      media_body=media_http.MediaFileUpload(
          options.file, chunksize=-1, resumable=True
      ),
  )

  resumable_upload(insert_request)


def parse_args() -> argparse.Namespace:
  """Parses command-line arguments for the upload sample."""
  parser = argparse.ArgumentParser(
      description='Upload a video to YouTube using the YouTube Data API v3.'
  )
  parser.add_argument('--file', required=True, help='Video file to upload.')
  parser.add_argument('--title', default='Test Title', help='Video title.')
  parser.add_argument(
      '--description', default='Test Description', help='Video description.'
  )
  parser.add_argument(
      '--category',
      default='22',
      help=(
          'Numeric video category. See '
          'https://developers.google.com/youtube/data/reference/rest/v3/videoCategories/list'
      ),
  )
  parser.add_argument(
      '--keywords', default='', help='Video keywords, comma separated.'
  )
  parser.add_argument(
      '--privacyStatus',
      choices=VALID_PRIVACY_STATUSES,
      default=VALID_PRIVACY_STATUSES[0],
      help='Video privacy status.',
  )
  return parser.parse_args()


def main() -> None:
  args = parse_args()
  if not os.path.exists(args.file):
    sys.exit('Specify a valid file using the --file= parameter.')

  youtube = get_authenticated_service()
  try:
    initialize_upload(youtube, args)
  except errors.HttpError as err:
    print(f'An HTTP error {err.resp.status} occurred:\n{err.content}')


if __name__ == '__main__':
  main()

Additional resources