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, andgoogle-auth-httplib2:pip install --upgrade google-api-python-client google-auth-oauthlib google-auth-httplib2OAuth 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.jsonin the same directory asupload_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.uploadscope, then caches the resulting access and refresh tokens intoken.jsonfor 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 toTest Title.Example: --title="Summer vacation in California"--description: Description of the video. Defaults toTest Description.Example: --description="Had fun surfing in Santa Cruz"--category: Numeric ID for the YouTube video category associated with the video. Defaults to22(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 arepublic,private, andunlisted. Defaults topublic; passprivateorunlistedwhen 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
- The Implement authorization guide covers OAuth 2.0 credentials, scopes, and token refresh flows.
- The Resumable Uploads guide details the underlying HTTP resumable upload protocol.
- The
videosandvideos.insertreference pages document all writable video metadata fields and upload parameters.