Merge pull request #8 from irachex/qiniu

Implement Qiniu storage

.gitignore

@@ -16,3 +16,4 @@ tmp/
 .tox
 .cache
 .coverage
+venv/

flask_storage/__init__.py

@@ -18,6 +18,7 @@
     'local': 'flask_storage.local.LocalStorage',
     's3': 'flask_storage.s3.S3Storage',
     'upyun': 'flask_storage.upyun.UpyunStorage',
+    'qiniu': 'flask_storage.qiniu.QiniuStorage',
 }
 
 

flask_storage/qiniu.py

@@ -0,0 +1,91 @@
+# coding: utf-8
+"""
+   flask_storage.qiniu
+    ~~~~~~~~~~~~~~~~~~~
+
+    Qiniu storage, save the file to Qiniu.
+"""
+
+from __future__ import absolute_import
+
+from werkzeug import cached_property
+import qiniu.rs
+import qiniu.io
+import qiniu.conf
+
+from ._base import BaseStorage
+from ._compat import urljoin
+from ._utils import ConfigItem
+
+
+_missing = object()
+
+
+class QiniuStorage(BaseStorage):
+
+    access_key = ConfigItem('access_key', required=True)
+    secret_key = ConfigItem('secret_key', required=True)
+
+    bucket = ConfigItem('bucket', required=True)
+    base_url = ConfigItem('base_url', default=_missing)
+    base_dir = ConfigItem('base_dir')
+
+    def __init__(self, name, extensions, config):
+        super(QiniuStorage, self).__init__(name, extensions, config)
+        self.init_config()
+
+    def init_config(self):
+        qiniu.conf.ACCESS_KEY = self.access_key
+        qiniu.conf.SECRET_KEY = self.secret_key
+
+    @cached_property
+    def _client(self):
+        return qiniu.rs.Client()
+
+    def url(self, filename):
+        """Generate the url for a filename.
+
+        :param filename: Name of the file.
+        """
+        if self.base_url is _missing:
+            base_url = 'http://%s.qiniudn.com/' % self.bucket
+        else:
+            base_url = self.base_url
+
+        if self.base_dir:
+            urlbase = urljoin(base_url, self.base_dir)
+
+        return urljoin(urlbase, filename)
+
+    def generate_upload_token(self, filename=None):
+        """
+        Generate a upload token used by client.
+        :param filename: Client can upload file if filename is None.
+                    Otherwise, client can modify the file.
+        """
+        if filename:
+            scope = '%s:%s' % (self.bucket, filename)
+        else:
+            scope = self.bucket
+        policy = qiniu.rs.PutPolicy(scope)
+        return policy.token()
+
+    def save(self, storage, filename, token=None, extra=None):
+        self.check(storage)
+        if token is None:
+            token = self.generate_upload_token()
+        stream = storage.stream
+        ret, err = qiniu.io.put(token, filename, stream, extra=extra)
+        if err:
+            raise QiniuException(err)
+        return ret
+
+    def delete(self, filename):
+        ret, err = self._client.delete(self.bucket, filename)
+        if err:
+            raise QiniuException(err)
+        return ret
+
+
+class QiniuException(Exception):
+    pass

tests/_base.py

@@ -0,0 +1,37 @@
+from mock import patch
+import unittest
+
+from flask import Flask, request
+
+
+class BaseCase(unittest.TestCase):
+    CONFIG = {
+        'TESTING': True,
+        'DEBUG': True,
+    }
+
+    def setUp(self):
+        self.app = self.create_app()
+        self.client = self.app.test_client()
+
+    def create_app(self):
+        app = Flask(__name__)
+        app.config.update(self.CONFIG)
+
+        @app.route('/upload', methods=['POST'])
+        def upload():
+            image = request.files.get('image')
+            return str(self.storage.save(image, 'flask.png'))
+
+        return app
+
+    def upload(self):
+        image = self.app.open_resource("flask.png")
+        response = self.client.post('/upload', data={'image': image})
+        return response
+
+    def patch(self, target, *args, **kwargs):
+        patcher = patch(target, *args, **kwargs)
+        patched_object = patcher.start()
+        self.addCleanup(patcher.stop)
+        return patched_object

tests/test_local.py

@@ -1,35 +1,8 @@
 import os
 import shutil
-import unittest
 
-from flask import Flask, request
 from flask_storage.local import LocalStorage
-
-
-class BaseCase(unittest.TestCase):
-    CONFIG = {
-        'TESTING': True
-    }
-
-    def setUp(self):
-        self.app = self.create_app()
-        self.client = self.app.test_client()
-
-    def create_app(self):
-        app = Flask(__name__)
-        app.config.update(self.CONFIG)
-
-        @app.route('/upload', methods=['POST'])
-        def upload():
-            image = request.files.get('image')
-            return self.storage.save(image, 'flask.png')
-
-        return app
-
-    def upload(self):
-        image = self.app.open_resource("flask.png")
-        response = self.client.post('/upload', data={'image': image})
-        return response
+from ._base import BaseCase
 
 
 class TestLocalStorage(BaseCase):

tests/test_qiniu.py

@@ -0,0 +1,24 @@
+from flask_storage.qiniu import QiniuStorage
+from ._base import BaseCase
+
+
+class TestQiniuStorage(BaseCase):
+    CONFIG = dict(
+        access_key='test_access_key',
+        secret_key='test_secret_key',
+        bucket='qiniu_test'
+    )
+
+    storage = QiniuStorage('qiniu', None, CONFIG)
+
+    def setUp(self):
+        super(TestQiniuStorage, self).setUp()
+        self.io = self.patch('qiniu.io')
+        self.upload_token = self.patch('qiniu.rs.PutPolicy.token')
+
+    def test_upload(self):
+        ret, err = {'key': 'flask.png'}, None
+        self.io.put.return_value = ret, err
+        response = self.upload()
+        assert self.upload_token.call_count == 1
+        assert self.io.put.call_count == 1