Reformatted with recent version of Black

filestorage/__init__.py

@@ -11,27 +11,27 @@
 
 def _read(rel_path):
     here = os.path.abspath(os.path.dirname(__file__))
-    with codecs.open(os.path.join(here, rel_path), 'r') as fp:
+    with codecs.open(os.path.join(here, rel_path), "r") as fp:
         return fp.read()
 
 
-__version__ = _read('VERSION').strip()
+__version__ = _read("VERSION").strip()
 
 # Instantiate the store singleton
 store = StorageContainer()
 
 
 __all__ = [
-    'store',
-    'StorageContainer',
-    'StorageHandlerBase',
-    'AsyncStorageHandlerBase',
-    'FileItem',
-    'FilterBase',
-    'AsyncFilterBase',
-    'exceptions',
-    'handlers',
-    'filters',
-    'pyramid_config',
-    'config_utils',
+    "store",
+    "StorageContainer",
+    "StorageHandlerBase",
+    "AsyncStorageHandlerBase",
+    "FileItem",
+    "FilterBase",
+    "AsyncFilterBase",
+    "exceptions",
+    "handlers",
+    "filters",
+    "pyramid_config",
+    "config_utils",
 ]

filestorage/config_utils.py

@@ -9,15 +9,15 @@
 
 def try_import(default_module: str, model: str):
     """Attempt to import the given name."""
-    module_name, _, cls_name = model.rpartition('.')
+    module_name, _, cls_name = model.rpartition(".")
     module_name = module_name or default_module
     try:
         module = importlib.import_module(module_name)
         cls = getattr(module, cls_name)
-    except (ImportError):
-        raise ValueError('module not installed')
+    except ImportError:
+        raise ValueError("module not installed")
     except AttributeError:
-        raise ValueError('bad class name')
+        raise ValueError("bad class name")
 
     return cls
 
@@ -32,14 +32,14 @@ def get_init_properties(cls, to_class=object) -> Set[str]:
     Returns a set of all parameters found.
     """
     result = set()
-    init = getattr(cls, '__init__', None)
+    init = getattr(cls, "__init__", None)
 
     if init is not None:
         for param in inspect.signature(init).parameters.values():
             if param.kind == param.VAR_KEYWORD:
                 # Ignore any **kwargs
                 continue
-            if param.name == 'self':
+            if param.name == "self":
                 continue
 
             result.add(param.name)
@@ -53,7 +53,7 @@ def get_init_properties(cls, to_class=object) -> Set[str]:
 def setup_from_settings(
     settings: Dict[str, str],
     store: StorageContainer,
-    key_prefix: str = 'store',
+    key_prefix: str = "store",
 ) -> bool:
     """Setup the provided store with the settings dictionary.
     Will only pay attention to keys that start with the key_prefix.
@@ -68,7 +68,7 @@ def setup_from_settings(
 
     # If there's configuration to be had, setup the store with it.
     if settings_dict:
-        setup_store(store, key_prefix, '', settings_dict)
+        setup_store(store, key_prefix, "", settings_dict)
         return True
     else:
         # Otherwise, assume that the store isn't going to be used now
@@ -82,26 +82,26 @@ def setup_store(
     """Setup a specific store to the given name in the settings_dict.
     key_prefix denotes where this name came from (for good error messages).
     """
-    name = name or ''
+    name = name or ""
     try:
-        handler_class_name = settings_dict['handler'][None]
+        handler_class_name = settings_dict["handler"][None]
     except KeyError:
         raise FilestorageConfigError(
-            f'Pyramid settings has no key for {key_prefix}{name}.handler'
+            f"Pyramid settings has no key for {key_prefix}{name}.handler"
         )
-    if handler_class_name.lower() == 'none':
+    if handler_class_name.lower() == "none":
         handler = None
     else:
-        handler = get_handler(key_prefix + name, settings_dict['handler'])
+        handler = get_handler(key_prefix + name, settings_dict["handler"])
 
-    settings_dict.pop('handler')
+    settings_dict.pop("handler")
 
     store.handler = handler
 
     # Setup any sub-store configuration
     for key, sub_config in settings_dict.items():
-        if key.startswith('[') and key.endswith(']'):
-            sub_store = key.lstrip('[').rstrip(']').strip('"').strip("'")
+        if key.startswith("[") and key.endswith("]"):
+            sub_store = key.lstrip("[").rstrip("]").strip('"').strip("'")
             setup_store(
                 store=store[sub_store],
                 key_prefix=key_prefix + key,
@@ -110,53 +110,51 @@ def setup_store(
             )
         else:
             raise FilestorageConfigError(
-                f'Pyramid settings unknown key {key_prefix}.{key}'
+                f"Pyramid settings unknown key {key_prefix}.{key}"
             )
 
 
 def get_handler(key_prefix: str, settings_dict: Dict) -> StorageHandlerBase:
-    name = f'{key_prefix}.handler'
+    name = f"{key_prefix}.handler"
     handler_name = settings_dict.pop(None)
     try:
-        handler_cls = try_import('filestorage.handlers', handler_name)
+        handler_cls = try_import("filestorage.handlers", handler_name)
     except ValueError:
-        raise FilestorageConfigError(f'Pyramid settings bad value for {name}')
+        raise FilestorageConfigError(f"Pyramid settings bad value for {name}")
 
     valid_args = get_init_properties(handler_cls, StorageHandlerBase)
 
     kwargs = {}
     for key, value in settings_dict.items():
         if key not in valid_args:
             maybe = difflib.get_close_matches(key, valid_args, 1)
-            maybe_txt = ''
+            maybe_txt = ""
             if maybe:
                 maybe_txt = f' Did you mean "{name}.{maybe[0]}"?'
             raise FilestorageConfigError(
                 f'Pyramid invalid setting "{name}.{key}". {maybe_txt}'
             )
-        if key == 'filters':
-            kwargs['filters'] = get_all_filters(name, value)
+        if key == "filters":
+            kwargs["filters"] = get_all_filters(name, value)
         else:
             kwargs[key] = decode_kwarg(value)
 
     try:
         return handler_cls(**kwargs)
     except Exception as err:
-        raise FilestorageConfigError(
-            f'Pyramid settings bad args for {name}: {err}'
-        )
+        raise FilestorageConfigError(f"Pyramid settings bad args for {name}: {err}")
 
 
 def get_all_filters(key_prefix: str, settings_dict: Dict) -> List[FilterBase]:
     """Get all the filters from within the settings_dict"""
     filters: List[Tuple[int, FilterBase]] = []
     for filter_ref, filter_dict in settings_dict.items():
-        filter_prefix = f'{key_prefix}.filters{filter_ref}'
+        filter_prefix = f"{key_prefix}.filters{filter_ref}"
         try:
-            filter_id = int(filter_ref.lstrip('[').rstrip(']'))
+            filter_id = int(filter_ref.lstrip("[").rstrip("]"))
         except Exception as err:
             raise FilestorageConfigError(
-                f'Pyramid settings bad key {key_prefix}{filter_ref}: {err}'
+                f"Pyramid settings bad key {key_prefix}{filter_ref}: {err}"
             )
         filters.append((filter_id, get_filter(filter_prefix, filter_dict)))
 
@@ -168,18 +166,16 @@ def get_filter(key_prefix: str, settings_dict: Dict) -> FilterBase:
     """Get a single filter from within the settings_dict"""
     filter_name = settings_dict.pop(None)
     try:
-        filter_cls = try_import('filestorage.filters', filter_name)
+        filter_cls = try_import("filestorage.filters", filter_name)
     except ValueError:
-        raise FilestorageConfigError(
-            f'Pyramid settings bad value for {key_prefix}'
-        )
+        raise FilestorageConfigError(f"Pyramid settings bad value for {key_prefix}")
 
     kwargs = {key: decode_kwarg(value) for key, value in settings_dict.items()}
     try:
         return filter_cls(**kwargs)
     except Exception as err:
         raise FilestorageConfigError(
-            f'Pyramid settings bad args for {key_prefix}: {err}'
+            f"Pyramid settings bad args for {key_prefix}: {err}"
         )
 
 
@@ -198,21 +194,19 @@ def decode_kwarg(value) -> Any:
         try:
             value = value.pop(None)
         except KeyError:
-            raise ValueError(f'decode_kwarg got an invalid dict: {value!r}')
+            raise ValueError(f"decode_kwarg got an invalid dict: {value!r}")
         return decode_kwarg(value)
 
     if not isinstance(value, str):
-        raise ValueError(f'decode_kwarg expected a str, got: {value!r}')
-    if (value.startswith('[') and value.endswith(']')) or (
-        value.startswith('{') and value.endswith('}')
+        raise ValueError(f"decode_kwarg expected a str, got: {value!r}")
+    if (value.startswith("[") and value.endswith("]")) or (
+        value.startswith("{") and value.endswith("}")
     ):
         # handle lists, sets and dicts
         try:
             return eval(value, {}, {})
         except Exception as err:
-            raise FilestorageConfigError(
-                f'Pyramid settings bad value {value}: {err}'
-            )
+            raise FilestorageConfigError(f"Pyramid settings bad value {value}: {err}")
 
     if value.isdigit():
         return int(value)
@@ -258,7 +252,7 @@ def get_keys_from(prefix: str, settings: Dict) -> Dict:
     """Get nested dicts from a dictionary of . separated keys"""
     result: Dict = {}
     for key, value in settings.items():
-        if key.startswith(f'{prefix}.') or key.startswith(f'{prefix}['):
+        if key.startswith(f"{prefix}.") or key.startswith(f"{prefix}["):
             set_nested_value(key, value, result)
 
     return result.get(prefix, {})
@@ -269,8 +263,8 @@ def set_nested_value(key: str, value: str, result: Dict) -> Dict:
     sub = result
     # Add a . to each [ to make the parsing delimiter consistent:
     # 'foo[0][1]' to 'foo.[0].[1]'
-    key = key.replace('[', '.[')
-    for part in key.split('.'):
+    key = key.replace("[", ".[")
+    for part in key.split("."):
         sub = sub.setdefault(part, {})
     sub[None] = value.strip()
     return result

filestorage/config_utils.pyi

@@ -16,12 +16,8 @@ def setup_from_settings(
 def setup_store(
     store: StorageContainer, key_prefix: str, name: str, settings_dict: Dict
 ) -> Any: ...
-def get_handler(
-    key_prefix: str, settings_dict: Dict
-) -> StorageHandlerBase: ...
-def get_all_filters(
-    key_prefix: str, settings_dict: Dict
-) -> List[FilterBase]: ...
+def get_handler(key_prefix: str, settings_dict: Dict) -> StorageHandlerBase: ...
+def get_all_filters(key_prefix: str, settings_dict: Dict) -> List[FilterBase]: ...
 def get_filter(key_prefix: str, settings_dict: Dict) -> FilterBase: ...
 def unquote(value: str) -> str: ...
 def decode_kwarg(value: Any) -> Any: ...

filestorage/file_item.py

@@ -6,7 +6,7 @@
 
 
 class SyncReader:
-    def __init__(self, item: 'FileItem'):
+    def __init__(self, item: "FileItem"):
         self.data = item.data
         self.filename = item.filename
         if self.data is not None:
@@ -20,14 +20,14 @@ def seek(self, offset: int, whence: int = 0) -> int:
 
     def read(self, size: int = -1) -> bytes:
         if self.data is None:
-            return b''
+            return b""
         return self._reader(size)
 
     closed = False
 
 
 class AsyncReader:
-    def __init__(self, item: 'FileItem'):
+    def __init__(self, item: "FileItem"):
         self.data = item.data
         self.filename = item.filename
         if self.data is not None:
@@ -41,7 +41,7 @@ async def seek(self, offset: int, whence: int = 0) -> int:
 
     async def read(self, size: int = -1) -> bytes:
         if self.data is None:
-            return b''
+            return b""
         return await self._reader(size)
 
     closed = False
@@ -53,21 +53,18 @@ class FileItem(NamedTuple):
     data: Optional[BinaryIO] = None
     media_type: Optional[str] = None  # Formerly known as MIME-type
 
-    def copy(self, **kwargs) -> 'FileItem':
-        filename = kwargs.get('filename', self.filename)
-        path = kwargs.get('path', self.path)
-        data = kwargs.get('data', self.data)
-        media_type = kwargs.get('media_type', self.media_type)
+    def copy(self, **kwargs) -> "FileItem":
+        filename = kwargs.get("filename", self.filename)
+        path = kwargs.get("path", self.path)
+        data = kwargs.get("data", self.data)
+        media_type = kwargs.get("media_type", self.media_type)
 
-        return FileItem(
-            filename=filename, path=path, data=data, media_type=media_type
-        )
+        return FileItem(filename=filename, path=path, data=data, media_type=media_type)
 
     def __repr__(self) -> str:
-        has_data = 'no data' if self.data is None else 'with data'
+        has_data = "no data" if self.data is None else "with data"
         return (
-            f'<FileItem filename:{self.filename!r} '
-            f'path:{self.path!r} {has_data}>'
+            f"<FileItem filename:{self.filename!r} " f"path:{self.path!r} {has_data}>"
         )
 
     @property
@@ -77,7 +74,7 @@ def has_data(self) -> bool:
     @property
     def url_path(self) -> str:
         """A relative URL path string for this path/filename"""
-        return '/'.join(self.path + (self.filename,))
+        return "/".join(self.path + (self.filename,))
 
     @property
     def fs_path(self) -> str:

filestorage/filter_base.py

@@ -25,8 +25,7 @@ async def async_call(self, item: FileItem) -> FileItem:
         """Apply the filter asynchronously"""
         if not self.async_ok:
             raise FilestorageConfigError(
-                f'The {self.__class__.__name__} filter cannot be used '
-                'asynchronously'
+                f"The {self.__class__.__name__} filter cannot be used " "asynchronously"
             )
 
         if iscoroutinefunction(self._apply):

filestorage/filters/__init__.py

@@ -1,4 +1,4 @@
 from .randomize_filename import RandomizeFilename
 from .valid_extensions import ValidateExtension
 
-__all__ = ['RandomizeFilename', 'ValidateExtension']
+__all__ = ["RandomizeFilename", "ValidateExtension"]

filestorage/filters/valid_extensions.py

@@ -13,9 +13,7 @@ class ValidateExtension(FilterBase):
     async_ok = True
 
     def __init__(self, extensions: List[str]):
-        self.extensions = set(
-            ext.lower().strip(os.path.extsep) for ext in extensions
-        )
+        self.extensions = set(ext.lower().strip(os.path.extsep) for ext in extensions)
 
     def extension_allowed(self, ext: str) -> bool:
         """Determine if the provided file extension is allowed."""