【Python3+GCS】pythonを使ってGCSにテキストデータをアップロードする

準備

・python用のクライアントライブラリをインストール

$ pip install --upgrade google-cloud-storage

・GCPアカウントとプロジェクト、GCSバケットは作ってあるものとする（普通に管理ページからGUIに従って作れば簡単）

・サービスアカウントキーを発行して、ローカルに適宜保存しておく（※）

・ローカルに保存したサービスアカウントキーのファイルパスを、環境変数にGOOGLE_APPLICATION_CREDENTIALSとして登録しておく（※）

※サービスアカウントキーを使わずに、コンソールからログインして接続することもできる。コンソールから

$ gcloud auth application-default login

しておいて、GOOGLE_APPLICATION_CREDENTIALSを設定せずにおくと、ログインにつかったユーザーアカウントに付与された権限でGCPリソースにアクセスできる。(自動でサービスアカウントキー的なものを作られてるみたいだけど、とりあえず環境変数の設定は不要)

基本の接続方法

文字列をそのままファイル（ブロブ）としてアップロードする場合、

from google.cloud import storage

def upload(bucket_name: str, blob_name: str, contents: str):
    storage_client = storage.Client()
    blob = self.bucket.blob(blob_name)
    blob.upload_from_string(contents)

これだけ。

クライアント作って、ブロブ作って、文字列をアップロード。おわり。かんたん。

なお、upload_from_string()には任意の引数でcontent_typeというのがあって、これを指定しないと"text/plain"が指定されるので、日本語（などの２バイト文字）を保存＆GUI上でプレビューしたい時はちょっとだけ注意。（プログラムで取得する場合は適切にエンコードすれば大丈夫だけど、GCSの管理画面上でプレビューすると文字化けしたりする）

GCS接続の為のクラスを作る

from google.cloud import storage

class GcsInterfase:
   def __init__(self, project_id :str, bucket_name: str) -> None:
       self.storage_client = storage.Client(project= project_id)
       self.bucket = self.storage_client.bucket(bucket_name)
   
   def upload(self, blob_name: str, contents: str):
       blob = self.bucket.blob(blob_name)
       try:
           blob.upload_from_string(contents, content_type="application/json")
           print(f"{blob_name}をアップロードしました")
       except Exception as e:
           raise ConnectionError(f"GcsInterFace::サーバーへの保存に失敗しました{e}")

こんな感じにしておくと制御しやすいんじゃないんでしょうかね……

Gcsへの接続を管理するクラスを作っておいて、イニシャライザで接続先を指定、アップロードメソッドでファイル名と中身を受け取るようになっています。

projectIDは指定しなくてもいい（サービスアカウントキーを発行したプロジェクトに繋がる）んですが、複数プロジェクトを跨いで開発してる時は明示的に指定して上げたほうが後で苦労しません。

ついでに、今回はjsonをアップしたかったのでcontent_typeを指定してみました。接続に失敗したときの例外処理も追加。（頑張ってクライアントライブラリのソース追ってみたけど、どんな例外が来る可能性があるのか突き止め切れなかったのでとりあえずまるっと捕まえるようにしてあります）