google-api-python-client 完全ガイド: Google API を Python で使いこなす

$ pip install --upgrade google-api-python-client google-auth-httplib2 google-auth-oauthlib

from googleapiclient.discovery import build
# APIキーを安全な方法で設定してください（環境変数など）
API_KEY = 'YOUR_API_KEY'
# API名とバージョンを指定
API_SERVICE_NAME = 'youtube'
API_VERSION = 'v3'
# サービスオブジェクトを構築
youtube = build(API_SERVICE_NAME, API_VERSION, developerKey=API_KEY)
# 例: 'python programming' で動画を検索
search_response = youtube.search().list( q='python programming', part='snippet', maxResults=5
).execute()
# 結果の表示 (簡略化)
for item in search_response.get('items', []): print(f"Title: {item['snippet']['title']}")

import os
import pickle
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
# アクセスを要求するスコープ (APIごとに異なる)
# 例: Google Drive の読み取り/書き込み権限
SCOPES = ['https://www.googleapis.com/auth/drive']
# ダウンロードしたクライアントシークレットファイル
CLIENT_SECRET_FILE = 'credentials.json'
# 認証情報を保存するファイル
TOKEN_PICKLE_FILE = 'token.pickle' # または 'token.json' など
def get_credentials(): creds = None # token.pickle ファイルが存在すれば、保存された認証情報を読み込む if os.path.exists(TOKEN_PICKLE_FILE): with open(TOKEN_PICKLE_FILE, 'rb') as token: creds = pickle.load(token) # 認証情報が存在しないか、無効な場合 if not creds or not creds.valid: # 認証情報が期限切れで、リフレッシュトークンが存在する場合 if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) # 認証情報がない場合、ユーザーに認証フローを開始してもらう else: flow = InstalledAppFlow.from_client_secrets_file( CLIENT_SECRET_FILE, SCOPES) # ローカルサーバーを起動して認証フローを実行 creds = flow.run_local_server(port=0) # 新しい認証情報を token.pickle に保存 with open(TOKEN_PICKLE_FILE, 'wb') as token: pickle.dump(creds, token) return creds
# 認証情報を取得
credentials = get_credentials()
# サービスオブジェクトを構築 (例: Google Drive API v3)
drive_service = build('drive', 'v3', credentials=credentials)
# API呼び出しの例: ファイルリストを取得
results = drive_service.files().list( pageSize=10, fields="nextPageToken, files(id, name)").execute()
items = results.get('files', [])
if not items: print('No files found.')
else: print('Files:') for item in items: print(f"{item['name']} ({item['id']})")

from google.oauth2 import service_account
from googleapiclient.discovery import build
# サービスアカウントキーファイルのパス
SERVICE_ACCOUNT_FILE = 'path/to/your/service_account.json'
# アクセスを要求するスコープ
SCOPES = ['https://www.googleapis.com/auth/calendar.readonly'] # 例: Google Calendar 読み取り権限
# サービスアカウントキーファイルから認証情報を読み込む
credentials = service_account.Credentials.from_service_account_file( SERVICE_ACCOUNT_FILE, scopes=SCOPES)
# サービスオブジェクトを構築 (例: Google Calendar API v3)
calendar_service = build('calendar', 'v3', credentials=credentials)
# API呼び出しの例: カレンダーリストを取得
calendar_list = calendar_service.calendarList().list().execute()
for calendar_list_entry in calendar_list['items']: print(calendar_list_entry['summary'])

try: # 認証情報を取得 (前章の get_credentials() 関数) credentials = get_credentials() # Drive API v3 のサービスオブジェクトを構築 drive_service = build('drive', 'v3', credentials=credentials) # ファイル一覧を取得するリクエストを作成 # pageSize: 1ページあたりの最大件数 # fields: レスポンスに含めるフィールドを指定 (効率化のため) request = drive_service.files().list( pageSize=10, fields="nextPageToken, files(id, name, mimeType, modifiedTime)" # 取得するフィールドを限定 ) # リクエストを実行し、レスポンスを取得 results = request.execute() # レスポンスからファイル情報を取得 items = results.get('files', []) if not items: print('No files found.') else: print('Files:') for item in items: print(f"Name: {item['name']}, ID: {item['id']}, Type: {item['mimeType']}, Modified: {item['modifiedTime']}") # 次のページのトークンがあれば表示 (ページネーション用) next_page_token = results.get('nextPageToken') if next_page_token: print(f"\nNext page token: {next_page_token}")
except HttpError as error: # API呼び出し時のエラーハンドリング print(f'An error occurred: {error}') # エラーレスポンスの詳細を表示 (デバッグ用) print(error.content)
except Exception as e: # その他の予期せぬエラー print(f'An unexpected error occurred: {e}')

from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload # ファイルアップロード用
from googleapiclient.errors import HttpError
# (get_credentials() 関数は前章参照)
SCOPES_DRIVE = ['https://www.googleapis.com/auth/drive'] # Drive用のスコープ
try: credentials = get_credentials(SCOPES_DRIVE) # スコープを指定して認証情報取得 drive_service = build('drive', 'v3', credentials=credentials) # 1. ファイル一覧表示 (前章の例参照) print("--- Fetching File List ---") results = drive_service.files().list( pageSize=5, fields="files(id, name)").execute() items = results.get('files', []) if items: for item in items: print(f"- {item['name']} ({item['id']})") else: print("No files found.") # 2. テキストファイルのアップロード print("\n--- Uploading a Text File ---") file_metadata = { 'name': 'my_report.txt', # 'parents': ['FOLDER_ID'] # 特定のフォルダにアップロードする場合 } media = MediaFileUpload('files/report.txt', # ローカルのファイルパス mimetype='text/plain', resumable=True) # 大きなファイルは resumable=True を推奨 file = drive_service.files().create(body=file_metadata, media_body=media, fields='id, name').execute() print(f"File uploaded: {file.get('name')} (ID: {file.get('id')})") # 3. ファイルのダウンロード (例: アップロードしたファイル) print("\n--- Downloading the File ---") file_id = file.get('id') request = drive_service.files().get_media(fileId=file_id) # download_path = f"downloaded_{file.get('name')}" # 保存先パス # fh = io.FileIO(download_path, 'wb') # downloader = MediaIoBaseDownload(fh, request) # done = False # while done is False: # status, done = downloader.next_chunk() # print(f"Download {int(status.progress() * 100)}%.") # print(f"File downloaded to {download_path}") # (ダウンロード部分は MediaIoBaseDownload を使うのが一般的ですが、ここでは簡略化のため省略) # 簡単な取得 (テキストファイルなど小さい場合) content = request.execute() print(f"File content (first 50 chars): {content[:50].decode('utf-8')}...") # 4. ファイルの削除 (注意して実行！) # print("\n--- Deleting the File ---") # drive_service.files().delete(fileId=file_id).execute() # print(f"File with ID: {file_id} deleted.")
except HttpError as error: print(f'An error occurred: {error}')
except Exception as e: print(f'An unexpected error occurred: {e}')

import datetime
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
# (get_credentials() 関数は前章参照)
SCOPES_CALENDAR = ['https://www.googleapis.com/auth/calendar'] # Calendar用のスコープ
try: credentials = get_credentials(SCOPES_CALENDAR) calendar_service = build('calendar', 'v3', credentials=credentials) # 1. 今後10件のイベントを取得 print("--- Fetching Upcoming 10 Events ---") now = datetime.datetime.utcnow().isoformat() + 'Z' # 'Z' indicates UTC time events_result = calendar_service.events().list( calendarId='primary', # 'primary' はメインカレンダー timeMin=now, maxResults=10, singleEvents=True, # 繰り返しイベントを展開 orderBy='startTime' ).execute() events = events_result.get('items', []) if not events: print('No upcoming events found.') else: for event in events: start = event['start'].get('dateTime', event['start'].get('date')) print(f"- {start} {event['summary']}") # 2. 新しいイベントの作成 print("\n--- Creating a New Event ---") event_body = { 'summary': 'Python API Test Event', 'location': 'Online', 'description': 'A test event created via Google Calendar API.', 'start': { 'dateTime': (datetime.datetime.utcnow() + datetime.timedelta(days=1)).isoformat() + 'Z', 'timeZone': 'UTC', }, 'end': { 'dateTime': (datetime.datetime.utcnow() + datetime.timedelta(days=1, hours=1)).isoformat() + 'Z', 'timeZone': 'UTC', }, # 'attendees': [ # {'email': 'attendee1@example.com'}, # ], 'reminders': { # 通知設定 'useDefault': False, 'overrides': [ {'method': 'popup', 'minutes': 10}, # 10分前にポップアップ ], }, } created_event = calendar_service.events().insert(calendarId='primary', body=event_body).execute() print(f"Event created: {created_event.get('summary')} (ID: {created_event.get('id')}) Link: {created_event.get('htmlLink')}") # 3. イベントの削除 (注意して実行！) # print("\n--- Deleting the Event ---") # event_id_to_delete = created_event.get('id') # calendar_service.events().delete(calendarId='primary', eventId=event_id_to_delete).execute() # print(f"Event with ID: {event_id_to_delete} deleted.")
except HttpError as error: print(f'An error occurred: {error}')
except Exception as e: print(f'An unexpected error occurred: {e}')

from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
# (get_credentials() 関数は前章参照)
# APIキーを使う場合は認証不要
# --- APIキーを使用する場合 ---
API_KEY = 'YOUR_YOUTUBE_API_KEY' # 安全に管理すること
YOUTUBE_API_SERVICE_NAME = 'youtube'
YOUTUBE_API_VERSION = 'v3'
try: youtube_service = build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, developerKey=API_KEY) # 1. 動画検索 print("--- Searching Videos (using API Key) ---") search_response = youtube_service.search().list( q='Google Cloud Platform', # 検索クエリ part='snippet', # 取得する情報パート (id, snippet, contentDetails, status, statistics など) type='video', # 検索対象タイプ (video, channel, playlist) maxResults=5 # 最大取得件数 ).execute() videos = search_response.get('items', []) if not videos: print('No videos found.') else: for video in videos: print(f"- Title: {video['snippet']['title']} (ID: {video['id']['videoId']})")
# --- OAuth 2.0 を使用する場合 (例: 自分のチャンネル情報取得) ---
# SCOPES_YOUTUBE = ['https://www.googleapis.com/auth/youtube.readonly'] # 読み取り権限の例
# try:
# credentials = get_credentials(SCOPES_YOUTUBE)
# youtube_service_oauth = build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, credentials=credentials)
# # 2. 自分のチャンネル情報を取得
# print("\n--- Fetching My Channel Info (using OAuth) ---")
# channels_response = youtube_service_oauth.channels().list(
# mine=True, # 認証ユーザー自身の情報を取得
# part='snippet,statistics'
# ).execute()
# channels = channels_response.get('items', [])
# if not channels:
# print("Could not find your channel.")
# else:
# channel = channels[0]
# print(f"- Channel Name: {channel['snippet']['title']}")
# print(f"- Subscribers: {channel['statistics']['subscriberCount']}")
# print(f"- Views: {channel['statistics']['viewCount']}")
except HttpError as error: print(f'An HTTP error {error.resp.status} occurred:\n{error.content}')
except Exception as e: print(f'An unexpected error occurred: {e}')

def list_all_drive_files(drive_service): """Google Drive 内の全てのファイル名と ID をリストアップする""" files = [] page_token = None while True: try: response = drive_service.files().list( q="mimeType!='application/vnd.google-apps.folder'", # フォルダを除外 spaces='drive', fields='nextPageToken, files(id, name)', pageToken=page_token # 前回のレスポンスのトークンを使用 ).execute() files.extend(response.get('files', [])) page_token = response.get('nextPageToken', None) # 次のページトークンがなければループ終了 if page_token is None: break print(f"Fetched a page, next page token: {page_token[:10]}...") # 進捗表示 except HttpError as error: print(f'An error occurred during pagination: {error}') break # エラーが発生したら中断 return files
# --- 実行例 ---
# credentials = get_credentials(SCOPES_DRIVE)
# drive_service = build('drive', 'v3', credentials=credentials)
# all_files = list_all_drive_files(drive_service)
# print(f"\nTotal files found: {len(all_files)}")
# for file in all_files[:10]: # 最初の10件を表示
# print(f"- {file.get('name')} ({file.get('id')})")

from googleapiclient.http import BatchHttpRequest
import json
# コールバック関数: 各レスポンスを処理する
def batch_callback(request_id, response, exception): if exception: # リクエスト固有のエラー処理 print(f"Error for request {request_id}: {exception}") else: # 成功したレスポンスの処理 print(f"Success for request {request_id}:") # print(json.dumps(response, indent=2)) # レスポンス内容を表示 if 'summary' in response: # Calendarイベントの場合 print(f" Event Summary: {response['summary']}") elif 'name' in response: # Driveファイルの場合 print(f" File Name: {response['name']}")
# --- 実行例 (Calendar API と Drive API のリクエストをバッチ処理) ---
# credentials_calendar = get_credentials(SCOPES_CALENDAR)
# calendar_service = build('calendar', 'v3', credentials=credentials_calendar)
# credentials_drive = get_credentials(SCOPES_DRIVE)
# drive_service = build('drive', 'v3', credentials=credentials_drive)
# バッチリクエストオブジェクトを作成
batch = BatchHttpRequest(callback=batch_callback)
# バッチにリクエストを追加 (例: 複数のイベント取得とファイル取得)
# イベントIDリスト (仮)
event_ids = ['event_id_1', 'event_id_2']
for i, event_id in enumerate(event_ids): batch.add(calendar_service.events().get(calendarId='primary', eventId=event_id), request_id=f'event-{i}')
# ファイルIDリスト (仮)
file_ids = ['file_id_1', 'file_id_2']
for i, file_id in enumerate(file_ids): batch.add(drive_service.files().get(fileId=file_id, fields='id, name'), request_id=f'file-{i}')
# バッチリクエストを実行
# batch.execute(http=http) # httpオブジェクトが必要 (通常 credentials.authorize(Http()) で取得)
# 注: google-auth ライブラリを使用している場合、http オブジェクトの取得方法が異なる場合がある
# 例: authorized_http = google_auth_httplib2.AuthorizedHttp(credentials, http=httplib2.Http())
# batch.execute(http=authorized_http)
print("Executing batch request...")
# (実行部分は環境に合わせて http オブジェクトを準備してください)

import time
from googleapiclient.errors import HttpError
def execute_with_retry(request, max_retries=5): """ 指数バックオフを用いたリトライ処理付きでリクエストを実行 """ for n in range(max_retries + 1): try: return request.execute() except HttpError as e: # リトライすべきエラーか判定 (例: 5xx サーバーエラー、429 クォータ超過) if e.resp.status in [429, 500, 502, 503, 504]: if n == max_retries: print(f"Max retries reached. Error: {e}") raise # 最大リトライ回数に達したらエラーを送出 else: sleep_time = (2 ** n) + random.random() # 指数バックオフ + ジッター print(f"Retryable error ({e.resp.status}) encountered. Retrying in {sleep_time:.2f} seconds...") time.sleep(sleep_time) else: # リトライすべきでないエラー (例: 400 Bad Request, 401 Unauthorized, 403 Forbidden (クォータ以外), 404 Not Found) print(f"Non-retryable error encountered: {e}") raise # エラーを送出 except Exception as e: print(f"An unexpected error occurred: {e}") raise # 予期せぬエラーはリトライしない
# --- 実行例 ---
# request = drive_service.files().list(...)
# try:
# results = execute_with_retry(request)
# # 結果の処理
# except Exception as e:
# # 最終的なエラー処理
# print(f"Failed to execute request after retries: {e}")