[Python] How to properly setup my OAuth for google photo API

I want to fetch all the folder and img contains under a user Google Photo via a python script running in console. I absolutely need this script to run without any redirection to browser or else because I plan to run it via cron task.

For that, I have plan to use OAuth, store the credentials.json and the token and roll the script each week (maybe with a way to get a refreshed token each time don't know if it's work since I have the problem below).

Firstly, is it possible to do something like that?

Next, I have enabled the Google Photos Library API, and created a OAuth 2.0 Client IDs with type "TV and limited input".

Created the script below but when I run it I keep running into "Access blocked: Authorization Error" error message, the last I have is "Missing required parameter: redirect_uri". But I can't make my head around it.

import os
import sys
import argparse
import pickle
import logging
import shutil
from tqdm import tqdm
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

# Google Photos API Scope
SCOPES = ['https://www.googleapis.com/auth/photoslibrary.readonly']

def setup_logging():
logging.basicConfig(
level=logging.INFO,
format='[%(asctime)s] %(levelname)s: %(message)s',
handlers=[
logging.StreamHandler(sys.stdout)
]
)

def parse_arguments():
parser = argparse.ArgumentParser(description="Google Photos Downloader")

parser.add_argument('--save-dir', type=str, default='photos', help='Directory to save downloaded photos')
parser.add_argument('--credentials', type=str, default='credentials.json', help='Path to Google API credentials file')
parser.add_argument('--token', type=str, default='token_photos.pkl', help='Path to token file')
parser.add_argument('--page-size', type=int, default=100, help='Number of media items per API call (max 100)')

return parser.parse_args()

def ensure_disk_space(required_bytes):
total, used, free = shutil.disk_usage('/')
if free < required_bytes:
logging.error(f"Insufficient disk space. Required: {required_bytes} bytes, Available: {free} bytes.")
sys.exit(1)

def authenticate(credentials_path, token_path):
creds = None
if os.path.exists(token_path):
with open(token_path, 'rb') as token:
creds = pickle.load(token)

if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
logging.info("Refreshing access token...")
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(credentials_path, SCOPES)
auth_url, _ = flow.authorization_url(prompt='consent')

print(f"\nPlease visit this URL to authorize the application:\n{auth_url}\n")
code = input("Enter the authorization code: ").strip()

flow.fetch_token(code=code)
creds = flow.credentials

with open(token_path, 'wb') as token:
pickle.dump(creds, token)
logging.info(f"Saved new token to {token_path}")

return creds

def get_photos_service(creds):
return build('photoslibrary', 'v1', credentials=creds, static_discovery=False)

def download_file(session, url, dest_path):
try:
with session.get(url, stream=True) as r:
r.raise_for_status()
with open(dest_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=8192):
f.write(chunk)
return True
except Exception as e:
logging.error(f"Failed to download {url}: {e}")
return False

def main():
setup_logging()
args = parse_arguments()

os.makedirs(args.save_dir, exist_ok=True)
creds = authenticate(args.credentials, args.token)
service = get_photos_service(creds)

downloaded_files = set(os.listdir(args.save_dir))
session = service._http

next_page_token = None
total_downloaded = 0

logging.info("Starting download of Google Photos media items...")

try:
while True:
results = service.mediaItems().list(
pageSize=args.page_size,
pageToken=next_page_token
).execute()

items = results.get('mediaItems', [])
if not items:
logging.info("No more media items found.")
break

for item in tqdm(items, desc="Processing media items", unit="item"):
filename = item.get('filename')
base_url = item.get('baseUrl')
mime_type = item.get('mimeType', '')

if not filename or not base_url:
logging.warning("Skipping item with missing data.")
continue

dest_path = os.path.join(args.save_dir, filename)

if filename in downloaded_files:
logging.debug(f"Skipping existing file: {filename}")
continue

is_video = mime_type.startswith('video/')
download_url = f"{base_url}=dv" if is_video else f"{base_url}=d"

if download_file(session, download_url, dest_path):
downloaded_files.add(filename)
total_downloaded += 1
else:
logging.warning(f"Failed to download {filename}")

next_page_token = results.get('nextPageToken')
if not next_page_token:
break

except HttpError as error:
logging.error(f"An error occurred: {error}")

logging.info(f"Download completed. Total files downloaded: {total_downloaded}")


if __name__ == '__main__':
main()

Continue reading...