Implement file-admin based on Azure Blob Storage

flask_admin/contrib/fileadmin/azure.py

+from __future__ import absolute_import
+from datetime import datetime
+from datetime import timedelta
+from time import sleep
+import os.path as op
+
+try:
+    from azure.storage.blob import BlobPermissions
+    from azure.storage.blob import BlockBlobService
+except ImportError:
+    BlobPermissions = BlockBlobService = None
+
+from flask import redirect
+
+from . import BaseFileAdmin
+
+
+class AzureStorage(object):
+    """
+        Storage object representing files on an Azure Storage container.
+
+        Usage::
+
+            from flask_admin.contrib.fileadmin import BaseFileAdmin
+            from flask_admin.contrib.fileadmin.azure import AzureStorage
+
+            class MyAzureAdmin(BaseFileAdmin):
+                # Configure your class however you like
+                pass
+
+            fileadmin_view = MyAzureAdmin(storage=AzureStorage(...))
+
+    """
+    _fakedir = '.dir'
+    _copy_poll_interval_seconds = 1
+    _send_file_lookback = timedelta(minutes=15)
+    _send_file_validity = timedelta(hours=1)
+    separator = '/'
+
+    def __init__(self, container_name, connection_string):
+        """
+            Constructor
+
+            :param container_name:
+                Name of the container that the files are on.
+
+            :param connection_string:
+                Azure Blob Storage Connection String
+        """
+
+        if not BlockBlobService:
+            raise ValueError('Could not import Azure Blob Storage SDK. '
+                             'You can install the SDK using '
+                             'pip install azure-storage-blob')
+
+        self._container_name = container_name
+        self._connection_string = connection_string
+        self.__client = None
+
+    @property
+    def _client(self):
+        if not self.__client:
+            self.__client = BlockBlobService(
+                connection_string=self._connection_string)
+            self.__client.create_container(
+                self._container_name, fail_on_exist=False)
+        return self.__client
+
+    @classmethod
+    def _get_blob_last_modified(cls, blob):
+        last_modified = blob.properties.last_modified
+        tzinfo = last_modified.tzinfo
+        epoch = last_modified - datetime(1970, 1, 1, tzinfo=tzinfo)
+        return epoch.total_seconds()
+
+    @classmethod
+    def _ensure_blob_path(cls, path):
+        if path is None:
+            return None
+
+        path_parts = path.split(op.sep)
+        return cls.separator.join(path_parts).lstrip(cls.separator)
+
+    def get_files(self, path, directory):
+        path = self._ensure_blob_path(path)
+        directory = self._ensure_blob_path(directory)
+
+        path_parts = path.split(self.separator) if path else []
+        num_path_parts = len(path_parts)
+        folders = set()
+        files = []
+
+        for blob in self._client.list_blobs(self._container_name, path):
+            blob_path_parts = blob.name.split(self.separator)
+            name = blob_path_parts.pop()
+
+            blob_is_file_at_current_level = blob_path_parts == path_parts
+            blob_is_directory_file = name == self._fakedir
+
+            if blob_is_file_at_current_level and not blob_is_directory_file:
+                rel_path = blob.name
+                is_dir = False
+                size = blob.properties.content_length
+                last_modified = self._get_blob_last_modified(blob)
+                files.append((name, rel_path, is_dir, size, last_modified))
+            else:
+                next_level_folder = blob_path_parts[:num_path_parts + 1]
+                folder_name = self.separator.join(next_level_folder)
+                folders.add(folder_name)
+
+        folders.discard(directory)
+        for folder in folders:
+            name = folder.split(self.separator)[-1]
+            rel_path = folder
+            is_dir = True
+            size = 0
+            last_modified = 0
+            files.append((name, rel_path, is_dir, size, last_modified))
+
+        return files
+
+    def is_dir(self, path):
+        path = self._ensure_blob_path(path)
+
+        num_blobs = 0
+        for blob in self._client.list_blobs(self._container_name, path):
+            blob_path_parts = blob.name.split(self.separator)
+            is_explicit_directory = blob_path_parts[-1] == self._fakedir
+            if is_explicit_directory:
+                return True
+
+            num_blobs += 1
+            path_cannot_be_leaf = num_blobs >= 2
+            if path_cannot_be_leaf:
+                return True
+
+        return False
+
+    def path_exists(self, path):
+        path = self._ensure_blob_path(path)
+
+        if path == self.get_base_path():
+            return True
+
+        try:
+            next(iter(self._client.list_blobs(self._container_name, path)))
+        except StopIteration:
+            return False
+        else:
+            return True
+
+    def get_base_path(self):
+        return ''
+
+    def get_breadcrumbs(self, path):
+        path = self._ensure_blob_path(path)
+
+        accumulator = []
+        breadcrumbs = []
+        for folder in path.split(self.separator):
+            accumulator.append(folder)
+            breadcrumbs.append((folder, self.separator.join(accumulator)))
+        return breadcrumbs
+
+    def send_file(self, file_path):
+        file_path = self._ensure_blob_path(file_path)
+
+        if not self._client.exists(self._container_name, file_path):
+            raise ValueError()
+
+        now = datetime.utcnow()
+        url = self._client.make_blob_url(self._container_name, file_path)
+        sas = self._client.generate_blob_shared_access_signature(
+            self._container_name, file_path,
+            BlobPermissions.READ,
+            expiry=now + self._send_file_validity,
+            start=now - self._send_file_lookback)
+        return redirect('%s?%s' % (url, sas))
+
+    def read_file(self, path):
+        path = self._ensure_blob_path(path)
+
+        blob = self._client.get_blob_to_bytes(self._container_name, path)
+        return blob.content
+
+    def write_file(self, path, content):
+        path = self._ensure_blob_path(path)
+
+        self._client.create_blob_from_text(self._container_name, path, content)
+
+    def save_file(self, path, file_data):
+        path = self._ensure_blob_path(path)
+
+        self._client.create_blob_from_stream(self._container_name, path,
+                                             file_data.stream)
+
+    def delete_tree(self, directory):
+        directory = self._ensure_blob_path(directory)
+
+        for blob in self._client.list_blobs(self._container_name, directory):
+            self._client.delete_blob(self._container_name, blob.name)
+
+    def delete_file(self, file_path):
+        file_path = self._ensure_blob_path(file_path)
+
+        self._client.delete_blob(self._container_name, file_path)
+
+    def make_dir(self, path, directory):
+        path = self._ensure_blob_path(path)
+        directory = self._ensure_blob_path(directory)
+
+        blob = self.separator.join([path, directory, self._fakedir])
+        blob = blob.lstrip(self.separator)
+        self._client.create_blob_from_text(self._container_name, blob, '')
+
+    def _copy_blob(self, src, dst):
+        src_url = self._client.make_blob_url(self._container_name, src)
+        copy = self._client.copy_blob(self._container_name, dst, src_url)
+        while copy.status != 'success':
+            sleep(self._copy_poll_interval_seconds)
+            copy = self._client.get_blob_properties(
+                self._container_name, dst).properties.copy
+
+    def _rename_file(self, src, dst):
+        self._copy_blob(src, dst)
+        self.delete_file(src)
+
+    def _rename_directory(self, src, dst):
+        for blob in self._client.list_blobs(self._container_name, src):
+            self._rename_file(blob.name, blob.name.replace(src, dst, 1))
+
+    def rename_path(self, src, dst):
+        src = self._ensure_blob_path(src)
+        dst = self._ensure_blob_path(dst)
+
+        if self.is_dir(src):
+            self._rename_directory(src, dst)
+        else:
+            self._rename_file(src, dst)
+
+
+class AzureFileAdmin(BaseFileAdmin):
+    """
+        Simple Azure Blob Storage file-management interface.
+
+            :param container_name:
+                Name of the container that the files are on.
+
+            :param connection_string:
+                Azure Blob Storage Connection String
+
+        Sample usage::
+
+            from flask_admin import Admin
+            from flask_admin.contrib.fileadmin.azure import AzureFileAdmin
+
+            admin = Admin()
+
+            admin.add_view(AzureFileAdmin('files_container', 'my-connection-string')
+    """
+
+    def __init__(self, container_name, connection_string, *args, **kwargs):
+        storage = AzureStorage(container_name, connection_string)
+        super(AzureFileAdmin, self).__init__(*args, storage=storage, **kwargs)

flask_admin/tests/fileadmin/test_fileadmin_azure.py

+import os.path as op
+from os import getenv
+from uuid import uuid4
+
+from nose import SkipTest
+
+from flask_admin.contrib.fileadmin import azure
+
+from .test_fileadmin import Base
+
+
+class AzureFileAdminTests(Base.FileAdminTests):
+    _test_storage = getenv('AZURE_STORAGE_CONNECTION_STRING')
+
+    def setUp(self):
+        if not azure.BlockBlobService:
+            raise SkipTest('AzureFileAdmin dependencies not installed')
+
+        self._container_name = 'fileadmin-tests-%s' % uuid4()
+
+        if not self._test_storage or not self._container_name:
+            raise SkipTest('AzureFileAdmin test credentials not set')
+
+        client = azure.BlockBlobService(connection_string=self._test_storage)
+        client.create_container(self._container_name)
+        dummy = op.join(self._test_files_root, 'dummy.txt')
+        client.create_blob_from_path(self._container_name, 'dummy.txt', dummy)
+
+    def tearDown(self):
+        client = azure.BlockBlobService(connection_string=self._test_storage)
+        client.delete_container(self._container_name)
+
+    def fileadmin_class(self):
+        return azure.AzureFileAdmin
+
+    def fileadmin_args(self):
+        return (self._container_name, self._test_storage), {}