PythonでGCSを使ってみた

やること

ベーシック認証を通してrequestsでデータ取得をし、CSVの形でGCSに格納する

前提

以前作成した下記記事の環境からスタート

GCSコンソール

GCSのコンソール上で「バケットを作成」
※今回は名前がtest
その下の階層に「フォルダを作成」
※こちらも名前がtest

コード

まずは、ローカルでCSVファイルを確認
requirements.txt

requests
google-cloud-storage

上記2つを追加、pip install -r requirements.txt

CSVファイルを格納しておくdownloadフォルダを作成しておく

app.py

import os
from flask import Flask
import requests
from requests.auth import HTTPBasicAuth
from google.cloud import storage

ID = "ooooooooo"
PASS = "xxxxxxxx"

app = Flask(__name__)

@app.route("/")
def test():
    
    # データ取得
    response = requests.get(
    "URL",  
    auth=HTTPBasicAuth(ID, PASS)
    )
    
    # csvの形になっているか確認
    with open("downloadフォルダ「パスのコピー」/test.csv", mode='w') as f:
        f.write(response.text)

    return "ok"

if __name__ == "__main__":
    app.run(debug=True, host='0.0.0.0', port=int(os.environ.get('PORT', 8080)))

• ターミナル1でpython3 app.py
• ターミナル2でcurl http://localhost:8080
  先ほど作成したdownloadフォルダにtest.csvとして格納されていればOK！

GCSに格納

確認用の# csvの形になっているか確認の部分は削除して、そこに下記を記入する

# Google Cloud StorageにCSVとして格納
client = storage.Client()
bucket = client.get_bucket("test")
blob = bucket.blob(f'test/test20220606.csv')
blob.upload_from_string(
    response.text,
    content_type='text/csv')

• ターミナル1でpython3 app.py
• ターミナル2でcurl http://localhost:8080
  GCSのコンソールで確認！
  バケット:testフォルダ名:testファイル名:test20220606.csv
  のように入っていればOK！
  ※念の為にダウンロードしてみると○

最後に

今回、初めてGCSを使ったので備忘録としても作成しました。
Cloud Schedulerと連動させると一定の時間に実行してくれるので便利です！
もし、Cloud Schedulerからデータを送る場合はPOSTにするので下記のように変更するだけで良いです。

@app.route("/",  methods=['POST'])

参考文献