事前準備

サービスアカウントのキーファイルの作成

OAuth用の認証キー(サービスアカウントキー)を用意する。 jsonファイルの例（一部マスクしてあります）

{
  "type": "service_account",
  "project_id": "＜プロジェクト名＞",
  "private_key_id": "****",
  "private_key": "-----BEGIN PRIVATE KEY-----\n 〜  \n-----END PRIVATE KEY-----\n",
  "client_email": "****@****.iam.gserviceaccount.com",
  "client_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_x509_cert_url": "https://www.googleapis.com/〜.iam.gserviceaccount.com"
}

※ 秘密鍵となるので、管理に気をつけましょう。外部に漏れてしまった場合はGCPのコンソールから無効化すること。

ライブラリのインストール

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

GoogleDrive

認証用メソッド

# get_drive_service.py

import json
import os
import pickle

from google.auth.transport.requests import Request

from google_auth_oauthlib.flow import InstalledAppFlow

from googleapiclient.discovery import build


DRIVE_SCOPES = ['https://www.googleapis.com/auth/drive']
DRIVE_API_SERVICE_NAME = 'drive'
DRIVE_API_VERSION = 'v3'
CREDENTIALS_DRIVE_PICKLE = "drive_credentials.pickle"


def get_drive_service():
    return get_authenticated_service(
        DRIVE_API_SERVICE_NAME,
        DRIVE_API_VERSION,
        pickle_file=CREDENTIALS_DRIVE_PICKLE,
        scopes=DRIVE_SCOPES
    )


def get_authenticated_service(api_service_name,
                              api_version,
                              pickle_file,
                              scopes=SCOPES):
    pickle_path = "%s/%s" % (CREDENTIALS_CACHE_DIR, pickle_file)
    credentials = get_pickled_credentials(pickle_file)
    if credentials:
        print("[%s] credential expiry(old): %s" % (pickle_file, credentials.expiry))
        req = Request()
        credentials.refresh(req)
        print("[%s] credential expiry(new): %s" % (pickle_file, credentials.expiry))
        with open(pickle_path, "wb") as f:
            pickle.dump(credentials, f)
        return build(api_service_name, api_version, credentials=credentials)

    info_str = os.environ.get("GOOGLE_SERVICE_ACCOUNT")
    if not info_str:
        raise EnvironmentError("Cannnot found env.")

    info = json.loads(info_str)
    flow = InstalledAppFlow.from_client_config(info, scopes=scopes)
    credentials = flow.run_console()
    with open(pickle_path, "wb") as f:
        pickle.dump(credentials, f)
    return build(api_service_name, api_version, credentials=credentials)

GDrive上にフォルダを作成する

mimeType: application/vnd.google-apps.folder にする

def create_folder(drive_service, name: str, parent_id: str):
    file_metadata = {
        'name': name,
        'mimeType': 'application/vnd.google-apps.folder',
        'parents': [parent_id]
    }
    file = drive_service.files().create(
        body=file_metadata,
    ).execute()
    return file

csvファイルをspreadsheetとしてuploadする

MediaIoBaseUpload の mimetypeは text/csv 、 bodyパラメタのmimeTypeは application/vnd.google-apps.spreadsheet にする

import io

from googleapiclient.http import MediaIoBaseUpload


def upload_csv_file(drive_service, filename: str, csv_data: str, parent_id: str):
    fh = io.BytesIO(csv_data.encode("utf-8"))
    media = MediaIoBaseUpload(fh, mimetype='text/csv')
    body = {
        "name": filename,
        "parents": [parent_id],
        'mimeType': 'application/vnd.google-apps.spreadsheet'
    }
    return drive_service.files().create(
        body=body, media_body=media
    ).execute()

ファイルを参照する

（みつけたら書きます。）

Youtube 情報の取得 (Data API, Report API)

（みつけたら書きます）

Google Cloud Storage バケットへテキストファイルをアップロードする

from google.cloud import storage
from google.oauth2 import service_account

GOOGLE_CLOUD_STORAGE_SCOPE_READWRITE = [
    "https://www.googleapis.com/auth/devstorage.read_write"
]
GSERVICEACCOUNT_FILEPATH = "./config/gcp/service-account-sample.json"


def get_storage_client():
    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = GSERVICEACCOUNT_FILEPATH
    credentials = service_account.Credentials.from_service_account_file(
        GSERVICEACCOUNT_FILEPATH,
        scopes=GOOGLE_CLOUD_STORAGE_SCOPE_READWRITE
    )
    client = storage.Client(credentials=credentials)
    return client


def upload_text_file(client, file_str, filename, content_type="text/plain"):
    bucket = client.bucket(GCS_BUCKET_NAME)
    blob = bucket.blob(filename)
    blob.upload_from_string(
        file_str,
        content_type=content_type)


if __name__ == "__main__":
    client = get_storage_client()
    upload_text_file(client, "this is test text", "sample.txt")