はじめに
Pythonを利用してGoogle Cloud StorageにPDFをアップロードやダウンロード、削除する手順です。
Pythonでpdfファイルを受け取るAPIサーバを構築した際に、Byte形式で受け取ったままPDFファイルをアップロードしようとして少し詰まったので備忘録です。
ソースコードの全体像
事前準備
Google Cloud Storageを利用するためにgoogle-cloud-storage
のライブラリをインストールする。
pip install google-cloud-storage
下記の手順を参考に認証情報を作成しsecret.json
として保存する。
実装
PDFファイルを操作するラッパークラスの作成
Google Cloud Storageを利用するためラッパークラスを定義する。
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()
バケット名と認証情報を引数とし、下記のように初期化する。
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
で指定したファイルをダウンロードする。
~~省略~~
# ダウンロード処理
download_byte = gcs.download(FILENAME)
with open("download.dpf", "wb") as f:
f.write(download_byte)
~~省略~~
削除
バケット上に存在する、FILENAME
で指定したファイルを削除する。
~~省略~~
# 削除
gcs.delete(FILENAME)
~~省略~~