はじめに

Pythonを利用してGoogle Cloud StorageにPDFをアップロードやダウンロード、削除する手順です。

Pythonでpdfファイルを受け取るAPIサーバを構築した際に、Byte形式で受け取ったままPDFファイルをアップロードしようとして少し詰まったので備忘録です。

ソースコードの全体像

下記のGistを参照

https://gist.github.com/sey323/1528ed17aff494078fa4fa5e5e0b6bd8

事前準備

Google Cloud Storageを利用するためにgoogle-cloud-storageのライブラリをインストールする。

pip install google-cloud-storage

下記の手順を参考に認証情報を作成しsecret.jsonとして保存する。

実装

PDFファイルを操作するラッパークラスの作成

Google Cloud Storageを利用するためラッパークラスを定義する。

gcp.py

from google.cloud import storage

class GoogleCloudStorage:
    def __init__(self, bucket_name:str,cred_path: str) -> None:
        _storage_client = storage.Client.from_service_account_json(cred_path)
        self._bucket = _storage_client.bucket(
            bucket_name 
        )
        print(
            f"Google Cloud Storage Initialize Complete!. bucket name: {self._bucket.name}"
        )

    def upload(self, data: bytes, destination_blob_name: str) -> str:
        blob = self._bucket.blob(destination_blob_name)
        generation_match_precondition = 0

        blob.upload_from_string(
            data,
            content_type="application/pdf", # ←content_typeを指定する必要あり。
            if_generation_match=generation_match_precondition,
        )

        print(f"File uploaded to {destination_blob_name}.")


    def download(self,destination_blob_name: str) -> bytes:
        blob = self._bucket.blob(destination_blob_name)
        print(f"File download from {destination_blob_name}.")

        return blob.download_as_bytes() # Byte形式でダウンロード


    def delete(self,destination_blob_name: str):
        blob = self._bucket.blob(destination_blob_name)
        print(f"Delete file, {destination_blob_name}.")
        if blob.exists():
            blob.delete()

バケット名と認証情報を引数とし、下記のように初期化する。

main.py

from gcp import GoogleCloudStorage

GOOGLE_CLOUD_STORAGE_BUCKET_NAME = "test" # 保存するバケットの名称、任意の名称を指定する。
CREDENTIAL_PATH="secret.json" # Google Cloud Storageの認証情報

gcs = GoogleCloudStorage(GOOGLE_CLOUD_STORAGE_BUCKET_NAME,CREDENTIAL_PATH)

~~省略~~

アップロード

適当なPDFファイルを用意してFILENAMEに指定する。

~~省略~~

FILENAME = "sample.pdf"

# アップロード処理
with open(FILENAME, 'rb') as f:
    binary = f.read()
    gcs.upload(binary, FILENAME)

~~省略~~

Google Cloud StorageのGOOGLE_CLOUD_STORAGE_BUCKET_NAMEに指定したバケットに、FILENAMEで指定した名称でファイルが保存される。

ダウンロード

バケット上に存在する、FILENAMEで指定したファイルをダウンロードする。

main.py

~~省略~~

# ダウンロード処理
download_byte = gcs.download(FILENAME)
with open("download.dpf", "wb") as f:
    f.write(download_byte)

~~省略~~

削除

バケット上に存在する、FILENAMEで指定したファイルを削除する。

main.py

~~省略~~

# 削除
gcs.delete(FILENAME)

~~省略~~