Why not login to Qiita and try out its useful features?

We'll deliver articles that match you.

You can read useful information later.

2
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

PythonでGoogle Cloud StorageにPDFをアップロード/ダウンロード/削除

Last updated at Posted at 2023-05-05

はじめに

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を利用するためラッパークラスを定義する。

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)

~~省略~~
2
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
2
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?