レンタルサーバ上にWordPressを構築する準備をしています。
で、バックアップしたファイルを、「Dropboxへファイル転送する」ことにしました。
前回のエントリーで、Dropbox SDKに付属しているExample(cli_client.pyというサンプル)を試してみました。
Dropbox API(Python)を使ってみる « Walk on apps.
今回は実際にバックアップで使うものを試作してみます。
使っている環境
FreeBSD8.1(64bit)
Python 2.6.5
dropbox-python-sdk-1.5.1
(1)はじめに
今回、僕がやろうとしているのは、Dropbox SDKに付属しているサンプルcli_client.pyを流用し、自分が行いたい処理の部分を作成するという感じです。
cli_client.pyは、CUIとして動作します。コマンドラインプロンプトでコマンドを実行することで、DropBoxに対する操作ができるようになっています。このままだと、バッチ処理で自動実行させることができないため、すこしソースコードをいじります。
また、cli_client.pyには、DropBoxに対する様々な処理が組み込まれていますが、僕の場合、使いたい処理が3つだけなので、その部分のみ使うことにします
僕が行いたい処理は、おもにこの3つです。
・DropBox上のファイル名の一覧を取得する
・DropBoxへファイルの転送(送信)を行う
・DropBox上のファイルの削除
(2)Access token, Access token secretの取得
まずログイン認証をおこなって、Access token, Access token secretを取得しておきます。
Dropbox SDKに付属しているサンプルcli_client.pyを使って取得します。そうすると、token_store.txt というファイルに取得した値が保存されます。このファイルを残しておき、そのまま使います。
くわしくは、前回のエントリーに記載しているので、そちらを参照してください。
(3)コーディング
とりあえず作成したコードをまず書いておきます。内容についてはそのあと解説。
APP_KEY, APP_SECRETについては、自分で取得したものに置き換えてください。
#!/usr/local/bin/python
# -*- coding: utf-8 -*-
import cmd
import locale
import os
import pprint
import shlex
import time
from dropbox import client, rest, session
# XXX Fill in your consumer key and secret below
# You can find these at http://www.dropbox.com/developers/apps
APP_KEY = '********'
APP_SECRET = '********'
ACCESS_TYPE = 'app_folder' # should be 'dropbox' or 'app_folder' as configured for your app
def command(login_required=True):
"""a decorator for handling authentication and exceptions"""
def decorate(f):
def wrapper(self, *args):
if login_required and not self.sess.is_linked():
self.stdout.write("Please 'login' to execute this command\n")
return
try:
return f(self, *args)
except TypeError, e:
#self.stdout.write(str(e) + '\n')
print str(e)
except rest.ErrorResponse, e:
msg = e.user_error_msg or str(e)
#self.stdout.write('Error: %s\n' % msg)
print ('Error: %s' % msg)
wrapper.__doc__ = f.__doc__
return wrapper
return decorate
class DropboxTerm():
def __init__(self, app_key, app_secret):
self.sess = StoredSession(app_key, app_secret, access_type=ACCESS_TYPE)
self.api_client = client.DropboxClient(self.sess)
self.current_path = ''
self.sess.load_creds()
def do_ls(self):
"""list files in current remote directory"""
resp = self.api_client.metadata(self.current_path)
if 'contents' in resp:
for f in resp['contents']:
name = os.path.basename(f['path'])
encoding = locale.getdefaultlocale()[1]
#self.stdout.write(('%s\n' % name).encode(encoding))
print ('Filename=%s' % name).encode(encoding)
#print ('FullFilename=%s' % f['path']).encode(encoding)
@command()
def do_cd(self, path):
"""change current working directory"""
if path == "..":
self.current_path = "/".join(self.current_path.split("/")[0:-1])
else:
self.current_path += "/" + path
@command()
def do_put(self, from_path, to_path):
"""
Copy local file to Dropbox
Examples:
Dropbox> put ~/test.txt dropbox-copy-test.txt
"""
from_file = open(os.path.expanduser(from_path), "rb")
self.api_client.put_file(self.current_path + "/" + to_path, from_file)
@command()
def do_rm(self, path):
"""delete a file or directory"""
self.api_client.file_delete(self.current_path + "/" + path)
class StoredSession(session.DropboxSession):
"""a wrapper around DropboxSession that stores a token to a file on disk"""
TOKEN_FILE = "token_store.txt"
def load_creds(self):
try:
stored_creds = open(self.TOKEN_FILE).read()
self.set_token(*stored_creds.split('|'))
print "[loaded access token]"
except IOError:
pass # don't worry if it's not there
def write_creds(self, token):
f = open(self.TOKEN_FILE, 'w')
f.write("|".join([token.key, token.secret]))
f.close()
def delete_creds(self):
os.unlink(self.TOKEN_FILE)
def link(self):
request_token = self.obtain_request_token()
url = self.build_authorize_url(request_token)
print "url:", url
print "Please authorize in the browser. After you're done, press enter."
raw_input()
self.obtain_access_token(request_token)
self.write_creds(self.token)
def unlink(self):
self.delete_creds()
session.DropboxSession.unlink(self)
def main():
if APP_KEY == '' or APP_SECRET == '':
exit("You need to set your APP_KEY and APP_SECRET!")
term = DropboxTerm(APP_KEY, APP_SECRET)
print '-' * 30 + '<ChangeDir>'
term.do_cd("TestDir")
term.do_ls()
print '-' * 30 + '<PutFile>'
term.do_put('put.txt','put.txt')
term.do_ls()
time.sleep(20)
print '-' * 30 + '<RemoveFile>'
term.do_rm('put.txt')
term.do_ls()
if __name__ == '__main__':
main()
処理の内容は、
・DropBoxにアクセスし、ディレクトリ移動した後、ファイル名の一覧を取得
・put.txtというファイルを転送し、ファイル名一覧を表示(そしてｓｌｅｅｐ20秒)
・put.txtというファイルを削除し、ファイル名一覧を表示
になっています。
実行すると、こんな感じになります。
※locale設定してないとエラーになるので、export LANG=ja_JP.UTF-8 を実行しておいてください。
[********]$ ./test_DropboxClient.py
[loaded access token]
------------------------------<ChangeDir>
Filename=TestFile1.txt
Filename=TestFile2.txt
Filename=ドキュメント.txt
------------------------------<PutFile>
Filename=put.txt
Filename=TestFile1.txt
Filename=TestFile2.txt
Filename=ドキュメント.txt
------------------------------<RemoveFile>
Filename=TestFile1.txt
Filename=TestFile2.txt
Filename=ドキュメント.txt
でコードの説明に入ります。
今回は、cli_client.pyをほぼそのまま流用しています。DropboxTermクラスの中で、必要な処理を行っています。
オブジェクト作成した後、各メソッドを呼び出して使えるようになっています。
term = DropboxTerm(APP_KEY, APP_SECRET)
： ：
term.do_cd("TestDir")
term.do_ls()
DropboxTermクラスの中には、do_***といった形でメソッドが用意されています。ここで、メソッドの前に@command()といった形でPythonのデコレータが記載されています。
これにより、do_***といったメソッドが実行されるときに、command()の処理が一緒に実行されます。
デコレータについては僕自身あまり詳しくないんだけれども、デコレータを使うことにより冗長な処理を何度も記載する必要がなくなり、ソースコードが見やすくシンプルになります。
def command(login_required=True): の処理の中にdef decorate(f): という部分があると思います。このfのところにdo_***メソッドの内容が組み込まれた形で、処理が実行されます。
do_***メソッドが呼び出されると、command()処理の中のfのところにdo_***の内容が組み込まれて実行される。今回の場合、return f(self, *args)のところでdo_***の処理が実行されています。
def command(login_required=True): """a decorator for handling authentication and exceptions""" def decorate(f): def wrapper(self, *args): if login_required and not self.sess.is_linked(): self.stdout.write("Please 'login' to execute this command\n") return try: return f(self, *args) except TypeError, e: #self.stdout.write(str(e) + '\n') print str(e) except rest.ErrorResponse, e: msg = e.user_error_msg or str(e) #self.stdout.write('Error: %s\n' % msg) print ('Error: %s' % msg) wrapper.__doc__ = f.__doc__ return wrapper return decorate
ではこのcommand()処理の中でなにをやっているかというと、
・すでにログイン認証済かどうかのチェック
を行っています。(self.sess.is_linked()の部分)
ログイン認証に関する処理はStoredSessionクラスに記載されています。
補足すると、DropboxTermクラスの初期設定部分で、StoredSessionクラスが呼び出され、ログイン認証の初期設定が行われています。ここで作成されたセッション情報が、sessオブジェクトに格納されており、そのsessオブジェクトをcommand()処理の中でも使っています。
class DropboxTerm(): def __init__(self, app_key, app_secret): self.sess = StoredSession(app_key, app_secret, access_type=ACCESS_TYPE) self.api_client = client.DropboxClient(self.sess) self.current_path = '' self.sess.load_creds() ： ：
とりあえずこれで自分のやりたかったことは実現できそうです。
実際は、ファイル名に「年月日」をつけておいて、過去3か月分のファイルを残しておくような形でバックアップするつもり。で、3か月以上たったら削除する、って感じ。
Walk on apps. 