【GCP】CloudStorageからファイルをダウンロードする有効期限付きリンク（署名つきURL）を発行する

「GCSに放り込んだファイルのダウンロードURLが欲しいだけなのに」と丸一日ネットの海を彷徨ったので、記録を残しておきます

やりたいこと

集計結果をCloudStorageへぽんぽん放り込んでおいて、それを他の人に見て貰いたいけど、いちいちアカウント取って貰ってユーザーの権限設定して……とかが面倒な時に、「こっからDLできっから！」ってURLを渡したい。
※当然ながら、くれぐれも秘匿性の高い情報に使っちゃダメだよ！

参考サイト

GCP(CloudStorage) で署名付き URL を作成する - Tech Tips

PythonでGoogle Cloud Storageの署名付きURLを作成する - Qiita

結論

from google.cloud import storage
from oauth2client.service_account import ServiceAccountCredentials
import base64
import urllib
import time
import datetime
API_ACCESS_ENDPOINT = 'https://storage.googleapis.com'

def createBlob(bucket_name,blob_path,string):
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    targetBlob = bucket.get_blob(blob_path)
    if targetBlob == None:
   　　　targetBlob = bucket.blob(blob_path)
　　          
    targetBlob.upload_from_string(string)
    return

def sign_url(bucket, bucket_object, method, expires_after_seconds):
   gcs_filename = '/%s/%s' % (bucket, bucket_object)
   content_md5, content_type = None, None
   credentials = ServiceAccountCredentials.from_json_keyfile_name("GCP_SERCIVE_ACCOUNT_API_KEY")
   google_access_id = credentials.service_account_email
   expiration = datetime.datetime.now() + datetime.timedelta(seconds=expires_after_seconds)
   expiration = int(time.mktime(expiration.timetuple()))
   signature_string = '\n'.join([
       method,
       content_md5 or '',
       content_type or '',
       str(expiration),
       gcs_filename])
   _, signature_bytes = credentials.sign_blob(signature_string)
   signature = base64.b64encode(signature_bytes)
   query_params = {'GoogleAccessId': google_access_id,
                   'Expires': str(expiration),
                   'Signature': signature}
   return '{endpoint}{resource}?{querystring}'.format(
       endpoint=API_ACCESS_ENDPOINT,
       resource=gcs_filename,
       querystring=urllib.parse.urlencode(query_params))
 
 if __name__ == "__main__":
     string = "これはテストファイルです"
     bucket_name = "YOUR_BUCKET"
     blob_path = "FOLDER/FILE_NAME.txt"
     method = "GET"
     expires_after_seconds = 300 # 秒
     # ファイル（ブロブ）を作る
     createBlob(bucket_name,blob_path,string)
     # 署名つきURLを取得する
     signed_url = sign_url(bucket, blob_path, method, expires_after_seconds)

この記事は参考サイトと結論の覚え書きみたいなものだけど、サンプルコードの解読すら手こずったので一応

せつめい

def createBlob(bucket_name,blob_path,string):
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    targetBlob = bucket.get_blob(blob_path)
    if targetBlob == None:
   　　　targetBlob = bucket.blob(blob_path)
　　          
    targetBlob.upload_from_string(string)
    return
ブロブを作る（同じ名前のブロブがあれば更新、無ければ作成する）ところ。

ここはブロブを作る（同じ名前のブロブがあれば更新、無ければ作成する）ところ。そんなに難しくない。

bucket_nameは保存先バケットの名前を文字列で渡す。
blob_pathはバケット名を除いてブロブへのフルパス（フォルダ名/ファイル名.拡張子）を渡す。
今回は.txtだけど、.csvでも何でも、中身に応じて。（今回はblob.upload_from_string使っているので、このサンプルをそのまま使うなら、ファイルの中身はstringで渡せるものに限る。）

def sign_url(bucket, bucket_object, method, expires_after_seconds):

ここは、参考サイトを参考にほぼそのままコピペさせていただいたので、詳しい説明は参考サイトの方をどうぞ。（正直）

GCP_SERCIVE_ACCOUNT_API_KEY　にはローカル保存してあるサービスアカウントキーファイルのパスを入れます。（詳細はサービスアカウントキーで検索だ）
これ、GCP上で動かすならServiceAccountCredentials.from_json_keyfile_nameじゃなくて自動でCredentials入る方法がある気がする。でも今回はローカルバッチを作っているのでそこまで調べてる時間がないやごめんね。
気になる方は調べてみてください。そして教えてください。

if __name__ == "__main__":
     string = "これはテストファイルです"
     bucket_name = "YOUR_BUCKET"
     blob_path = "FOLDER/FILE_NAME.txt"
     method = "GET"
     expires_after_seconds = 300 # 秒
     # ファイル（ブロブ）を作る
     createBlob(bucket_name,blob_path,string)
     # 署名つきURLを取得する
     signed_url = sign_url(bucket, blob_path, method, expires_after_seconds)

動かす部分。

stringはファイルの中身。実際はごにゃごにゃ処理をしてきた結果とかが入ればいいです。
bucket_name はバケットの名前。
blob_path はブロブのパス（バケット名は不要、ファイル名.拡張子は必要）。
methodをputとかにすればURL経由で更新できたりすんのかな、と思いますが今回は不要なので触れません。
expires_after_seconds が、有効期限（秒単位）。300秒なら5分ですね。

URLを発行してから有効期限の間だけアクセスできますので、誰でもかれでも好き放題見えちゃうのはちょっと困るな～って場合でも安心。（ただしURLさえ解っていれば誰でも見れちゃうので、個人情報とかはこの形式でアクセスさせちゃだめだぞ！）

有効期限を2週間とかにしたい場合、60秒×60分×24時間×14日=1209600秒です。