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}")