LambdaでDropboxにファイルアップロード #Python

データバックアップをcsv形式で外部ストレージに

例えばRDSやDynamoDB上のデータを定期的に抽出して外部ストレージに置いておきたい、しかもS3ではなくDropboxに保存したいというケースがあったとします。
そこで、LambdaからDropbox API v2を呼ぶ形で実装します。

以前書いたPHPでDropbox APIを使ってアップロードする記事の応用となります。

利用するLambdaランタイム

Python3.8

1. csvファイルを出力する

まずはデータベースから抽出したデータを整形してcsvとして出力する部分です。
csvライブラリでさくっと。

    import csv
    #import os
    
    file = '/tmp/ファイル名.csv'
    # 環境変数で管理するのもよいです
    # file = '/tmp/' + os.environ['FILENAME']
    
    cols = ['項目名1', '項目名2', '項目名3', ...]
    
    with open(file, 'w') as f:
        writer = csv.DictWriter(f, fieldnames=cols)
        writer.writeheader()
    
        for row in rows: # DBから読み込んだデータ配列
            writer.writerow({ \
                '項目名1': row['col_1'], \
                '項目名2': row['col_2'], \
                '項目名3': row['col_3'], \
                ... \
            })

ポイントは、出力するファイルは/tmp/以下を指定することです。
そうしないと書き込み権限がないというエラーが発生します。
（参考）aws lambda file書き込む時のエラー

2. ファイルをDropboxにアップロード

こちらの記事で追記したように、Dropbox APIは短期的なアクセストークンは都度取得し直す必要があります。
予めリフレッシュトークンを取得しておいてください。
（参考）Dropbox API: アプリの権限とアクセストークンの移行

2-1. アクセストークン取得

PythonはDropbox公式SDKがあるので、予めローカルでインストールしてパッケージとしてzipファイルをLambdaにアップすればよいのですが、なるべく余計なライブラリを追加せずにやりたかったので、Lambda標準のurllib3でAPIエンドポイントを叩いていく方法でいきます。
（参考）Dropbox API v2 Document#/oauth2/token

    import os
    import urllib3
    
    # Dropbox API用の APP_KEY, APP_SECRET, REFRESH_TOKEN は環境変数管理がよいでしょう
    dbx_app_key = os.environ['DROPBOX_APP_KEY']
    dbx_app_secret = os.environ['DROPBOX_APP_SECRET']
    dbx_refresh_token = os.environ['DROPBOX_REFRESH_TOKEN']
    
    http = urllib3.PoolManager()
    
    # アクセストークン取得エンドポイント
    url = 'https://api.dropboxapi.com/oauth2/token'
    # APP_KEY,APP_SECRETはBasic認証のID・パスワードとして
    header = urllib3.util.make_headers(basic_auth='{}:{}'.format(dbx_app_key, dbx_app_secret))
    # リクエストパラメータ
    data = { \
        'grant_type': 'refresh_token', \
        'refresh_token': dbx_refresh_token \
    }
    
    resp = http.request('POST', url, headers=header, fields=data)
    
    if resp.status == 200:
        res = json.loads(resp.data)
        dbx_access_token = res['access_token'] # アクセストークン取得

エラー処理は省略しています。

2-2. ファイルアップロード

そして取得したアクセストークンを使ってアップロードです。

    import json
    import urllib3
    #import os
    
    # ファイルアップロードエンドポイント
    url = 'https://content.dropboxapi.com/2/files/upload'
    
    # Dropbox上のパス "/"始まりで記載
    dbx_up_path = '/バックアップフォルダ/バックアップファイル名.csv'
    # 環境変数で管理するのもよいです
    # dbx_up_path = os.environ['DROPBOX_DIR'] + os.environ['FILENAME']
    
    # ヘッダーの"Dropbox-API-Arg"パラメータにJSON形式でパスやその他オプションを指定
    dbx_param = { \
        'path': dbx_up_path, \
        'mode': { \
            '.tag': 'overwrite' \ # 上書きモード
        }
    }
    
    header = { \
        'Authorization': 'Bearer ' + dbx_access_token, \ # 2-1.で取得したアクセストークン
        'Content-Type': 'application/octet-stream', \
        'Dropbox-API-Arg': json.dumps(dbx_param) \
    }
    
    with open(file, 'rb') as fp: # 1.で出力したファイルパス
        bin = fp.read()
    
        # ファイルバイナリデータをリクエストパラメータとして
        resp = http.request('POST', url, headers=header, body=bin) # コネクションプール(http)は2-1.で取得したもの
    
        if resp.status == 200:
            print('SUCCESS!')

これで実際、Dropbox上にファイルがアップされていれば成功です。

おまけ

応用として、複数のcsvファイルを出力しておいて順次アップロードという形もできます。
（参考）Python でディレクトリ内のファイルをループする
以下、主に変更部分だけ。

    from pathlib import Path
    
    pathlist = Path('/tmp/').glob('*.csv')
    for path in pathlist:
    ...
        dbx_param = { \
            'path': '/バックアップフォルダ/' + path.name, \ # ファイル名
            'mode': { \
                '.tag': 'overwrite' \
            }
        }
    ...
        with open('/tmp/' + path.name, 'rb') as fp: # ファイルパス
            bin = fp.read()
    ...

以上です。
urllib3の使い方にやたらと四苦八苦したのですが、終わってみれば割とシンプルでした…