Source code for

# Copyright 2020 Karlsruhe Institute of Technology
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
import os

from flask import current_app
from flask import send_file
from flask_babel import lazy_gettext as _l

import kadi.lib.constants as const

from .core import BaseStorage

[docs]class LocalStorage(BaseStorage): """Storage provider that uses the local file system. :param root_directory: The directory the storage provider operates in. Must be an absolute path. :param num_dirs: (optional) Number of directories for local file paths generated by this storage provider. Must be a minimum of ``0``. :param dir_len: (optional) Length of each directory for local file paths generated by this storage provider. Must be a minimum of ``1``. :raises ValueError: If the given root directory is not suitable. """ def __init__(self, root_directory, num_dirs=3, dir_len=2): super().__init__(const.STORAGE_TYPE_LOCAL, storage_name=_l("Local")) if not os.path.isabs(root_directory): raise ValueError( f"Given root directory '{root_directory}' is not an absolute path." ) self._root_directory = root_directory self._num_dirs = max(num_dirs, 0) self._dir_len = max(dir_len, 1) @property def root_directory(self): """Get the root directory of this storage.""" return self._root_directory def _path_from_identifier(self, identifier): if self._num_dirs == 0: return identifier if len(identifier) <= self._dir_len * self._num_dirs: return None dirs = [ identifier[i : i + self._dir_len] for i in range(0, len(identifier), self._dir_len) ] filepath = os.path.join( *dirs[0 : self._num_dirs], identifier[self._num_dirs * self._dir_len :] ) return filepath def _create_filepath(self, identifier): error_msg = ( f"Given file identifier '{identifier}' is not suitable for creating a local" " file path." ) if not identifier or os.sep in identifier: raise ValueError(error_msg) filepath = self._path_from_identifier(identifier) if filepath is None: raise ValueError(error_msg) return os.path.join(self._root_directory, filepath) def _remove_empty_parent_dirs(self, filepath): current_dir = os.path.dirname(filepath) while ( os.path.isdir(current_dir) and not os.listdir(current_dir) and not os.path.samefile(self._root_directory, current_dir) ): try: os.rmdir(current_dir) current_dir = os.path.dirname(current_dir) except OSError: break
[docs] def exists(self, identifier): filepath = self._create_filepath(identifier) return os.path.isfile(filepath)
[docs] def get_size(self, identifier): filepath = self._create_filepath(identifier) return os.path.getsize(filepath)
[docs] def get_mimetype(self, identifier): file_size = self.get_size(identifier) with as f: return self._get_mimetype(f, file_size)
[docs] def open(self, identifier, mode="rb", encoding=None): filepath = self._create_filepath(identifier) os.makedirs(os.path.dirname(filepath), exist_ok=True) # pylint: disable=consider-using-with return open(filepath, mode=mode, encoding=encoding)
[docs] def save(self, identifier, stream, max_size=None): with, mode="wb") as f: return self._save(f, stream, max_size=max_size, calculate_checksum=True)
[docs] def move(self, src_identifier, dst_identifier): src_filepath = self._create_filepath(src_identifier) dst_filepath = self._create_filepath(dst_identifier) os.makedirs(os.path.dirname(dst_filepath), exist_ok=True) os.replace(src_filepath, dst_filepath) self._remove_empty_parent_dirs(src_filepath)
[docs] def delete(self, identifier): filepath = self._create_filepath(identifier) try: os.remove(filepath) except FileNotFoundError: pass self._remove_empty_parent_dirs(filepath)
[docs] def merge(self, identifier, identifier_list): with, mode="wb") as f: for chunk_identifier in identifier_list: with as f_chunk: self._save(f, f_chunk)
[docs] def download(self, identifier, *, filename, mimetype, as_attachment=True): filepath = self._create_filepath(identifier) response = send_file( filepath, download_name=filename, mimetype=mimetype, as_attachment=as_attachment, # In production environments, the web server handles conditional/range # responses via "X-Sendfile". conditional=current_app.environment != const.ENV_PRODUCTION, ) response.headers["Accept-Ranges"] = "bytes" return response