Skip to content

Commit

Permalink
Add type annotations
Browse files Browse the repository at this point in the history
modified:   CHANGES.rst
modified:   src/flask_uploads/__init__.py
modified:   src/flask_uploads/backwards_compatibility.py
modified:   src/flask_uploads/extensions.py
modified:   src/flask_uploads/flask_uploads.py
modified:   src/flask_uploads/test_helper.py
modified:   tox.ini
  • Loading branch information
jugmac00 committed Jul 4, 2020
1 parent 1f200cd commit ba3ad70
Show file tree
Hide file tree
Showing 7 changed files with 122 additions and 47 deletions.
1 change: 1 addition & 0 deletions CHANGES.rst
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ Changelog

unreleased
----------
- add type annotations
- drop support for Python 2 and Python 3.5
(`#8 <https://github.com/jugmac00/flask-reuploaded/issues/8>`_)
- deprecate `patch_request_class`
Expand Down
6 changes: 3 additions & 3 deletions src/flask_uploads/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,15 @@
from .extensions import SOURCE
from .extensions import EXECUTABLES
from .extensions import DEFAULTS
from .extensions import extension
from .extensions import lowercase_ext


from .flask_uploads import UploadConfiguration
from .flask_uploads import UploadSet
from .flask_uploads import addslash
from .flask_uploads import configure_uploads
from .flask_uploads import extension
from .flask_uploads import lowercase_ext
from .flask_uploads import config_for_set

from .test_helper import TestingFileStorage

__all__ = [
Expand Down
5 changes: 4 additions & 1 deletion src/flask_uploads/backwards_compatibility.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,10 @@
import warnings

from flask import Flask

def patch_request_class(app, size=64 * 1024 * 1024): # pragma: no cover

def patch_request_class(
app: Flask, size: int = 64 * 1024 * 1024) -> None: # pragma: no cover
"""Attention!
This function is deprecated and due to removal in `Flask-Reuploaded 1.0`.
Expand Down
11 changes: 6 additions & 5 deletions src/flask_uploads/extensions.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
"""Extension presets and extension configuration."""
import os
from typing import Iterable

# This contains archive and compression formats (.gz, .bz2, .zip, .tar,
# .tgz, .txz, and .7z).
Expand Down Expand Up @@ -67,7 +68,7 @@ class All:
This type can be used to allow all extensions. There is a predefined
instance named `ALL`.
"""
def __contains__(self, item):
def __contains__(self, item: str) -> bool:
return True


Expand All @@ -88,22 +89,22 @@ class AllExcept:
AllExcept(SCRIPTS + EXECUTABLES)
"""
def __init__(self, items):
def __init__(self, items: Iterable[str]) -> None:
self.items = items

def __contains__(self, item):
def __contains__(self, item: str) -> bool:
return item not in self.items


def extension(filename):
def extension(filename: str) -> str:
ext = os.path.splitext(filename)[1]
if ext.startswith('.'):
# os.path.splitext retains . separator
ext = ext[1:]
return ext


def lowercase_ext(filename):
def lowercase_ext(filename: str) -> str:
"""
This is a helper used by UploadSet.save to provide lowercase extensions for
all processed files, to compare with configured extensions in the same
Expand Down
98 changes: 72 additions & 26 deletions src/flask_uploads/flask_uploads.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,16 @@
import os
import os.path
import posixpath
from typing import Any
from typing import Callable
from typing import Dict
from typing import Iterable
from typing import Optional
from typing import Tuple
from typing import Union

from flask import Blueprint
from flask import Flask
from flask import abort
from flask import current_app
from flask import send_from_directory
Expand All @@ -25,13 +33,17 @@
from .extensions import lowercase_ext


def addslash(url):
def addslash(url: str) -> str:
if url.endswith('/'):
return url
return url + '/'


def config_for_set(uset, app, defaults=None):
def config_for_set(
uset: 'UploadSet',
app: Flask,
defaults: Optional[Dict[str, Optional[str]]] = None
) -> 'UploadConfiguration':
"""
This is a helper function for `configure_uploads` that extracts the
configuration for a single set.
Expand All @@ -48,8 +60,10 @@ def config_for_set(uset, app, defaults=None):
if defaults is None:
defaults = dict(dest=None, url=None)

allow_extensions = tuple(config.get(prefix + 'ALLOW', ()))
deny_extensions = tuple(config.get(prefix + 'DENY', ()))
allow_extensions = tuple(
config.get(prefix + 'ALLOW', ())) # Union[Tuple[()], Tuple[str, ...]]
deny_extensions = tuple(
config.get(prefix + 'DENY', ())) # Union[Tuple[()], Tuple[str, ...]]
destination = config.get(prefix + 'DEST')
base_url = config.get(prefix + 'URL')

Expand All @@ -66,14 +80,15 @@ def config_for_set(uset, app, defaults=None):
else:
raise RuntimeError("no destination for set %s" % uset.name)

if base_url is None and using_defaults and defaults['url']:
base_url = addslash(defaults['url']) + uset.name + '/'
if base_url is None and using_defaults:
if defaults['url'] is not None:
base_url = addslash(defaults['url']) + uset.name + '/'

return UploadConfiguration(
destination, base_url, allow_extensions, deny_extensions)


def configure_uploads(app, upload_sets):
def configure_uploads(app: Flask, upload_sets: Iterable['UploadSet']) -> None:
"""
Call this after the app has been configured. It will go through all the
upload sets, get their configuration, and store the configuration on the
Expand All @@ -93,8 +108,10 @@ def configure_uploads(app, upload_sets):
if not hasattr(app, 'upload_set_config'):
app.upload_set_config = {}
set_config = app.upload_set_config
defaults = dict(dest=app.config.get('UPLOADS_DEFAULT_DEST'),
url=app.config.get('UPLOADS_DEFAULT_URL'))
defaults = dict(
dest=app.config.get('UPLOADS_DEFAULT_DEST'),
url=app.config.get('UPLOADS_DEFAULT_URL')
) # type: Dict[str, Optional[str]]

for uset in upload_sets:
config = config_for_set(uset, app, defaults)
Expand All @@ -119,17 +136,29 @@ class UploadConfiguration:
:param deny: A list of extensions to deny, even if they are in the
`UploadSet` extensions list.
"""
def __init__(self, destination, base_url=None, allow=(), deny=()):
def __init__(
self,
destination: str,
base_url: Optional[str] = None,
allow: Union[Tuple[()], Tuple[str, ...]] = (),
deny: Union[Tuple[()], Tuple[str, ...]] = ()
) -> None:
self.destination = destination
self.base_url = base_url
self.allow = allow
self.deny = deny

@property
def tuple(self):
def tuple(self) -> Tuple[
str, Optional[str],
Union[Tuple[()], Tuple[str, ...]],
Union[Tuple[()], Tuple[str, ...]]
]:
return (self.destination, self.base_url, self.allow, self.deny)

def __eq__(self, other):
def __eq__(self, other: object) -> bool:
if not isinstance(other, UploadConfiguration):
return NotImplemented
return self.tuple == other.tuple


Expand All @@ -153,7 +182,12 @@ class UploadSet:
with the app, it should return the default upload
destination path for that app.
"""
def __init__(self, name='files', extensions=DEFAULTS, default_dest=None):
def __init__(
self,
name: str = 'files',
extensions: Iterable[str] = DEFAULTS,
default_dest: Optional[Callable[[Flask], str]] = None
) -> None:
if not name.isalnum():
raise ValueError("Name must be alphanumeric (no underscores)")
self.name = name
Expand All @@ -162,7 +196,7 @@ def __init__(self, name='files', extensions=DEFAULTS, default_dest=None):
self.default_dest = default_dest

@property
def config(self):
def config(self) -> 'UploadConfiguration':
"""
This gets the current configuration. By default, it looks up the
current application and gets the configuration from there. But if you
Expand All @@ -174,11 +208,14 @@ def config(self):
if self._config is not None:
return self._config
try:
return current_app.upload_set_config[self.name]
upload_configuration = (
current_app.upload_set_config[self.name]
) # type: UploadConfiguration
return upload_configuration
except AttributeError:
raise RuntimeError("cannot access configuration outside request")

def url(self, filename):
def url(self, filename: str) -> str:
"""
This function gets the URL a file uploaded to this set would be
accessed at. It doesn't check whether said file exists.
Expand All @@ -192,7 +229,7 @@ def url(self, filename):
else:
return base + filename

def path(self, filename, folder=None):
def path(self, filename: str, folder: Optional[str] = None) -> str:
"""
This returns the absolute path of a file uploaded to this set. It
doesn't actually check whether said file exists.
Expand All @@ -207,7 +244,7 @@ def path(self, filename, folder=None):
target_folder = self.config.destination
return os.path.join(target_folder, filename)

def file_allowed(self, storage, basename):
def file_allowed(self, storage: FileStorage, basename: str) -> bool:
"""This tells whether a file is allowed.
It should return `True` if the given
Expand All @@ -222,7 +259,7 @@ def file_allowed(self, storage, basename):
"""
return self.extension_allowed(extension(basename))

def extension_allowed(self, ext):
def extension_allowed(self, ext: str) -> bool:
"""
This determines whether a specific extension is allowed. It is called
by `file_allowed`, so if you override that but still want to check
Expand All @@ -233,10 +270,18 @@ def extension_allowed(self, ext):
return ((ext in self.config.allow) or
(ext in self.extensions and ext not in self.config.deny))

def get_basename(self, filename):
return lowercase_ext(secure_filename(filename))

def save(self, storage, folder=None, name=None):
def get_basename(self, filename: str) -> str:
# `secure_filename` is already typed via typeshed
# cf https://github.com/python/typeshed/pull/4308
# but not yet available via `mypy` from PyPi
return lowercase_ext(secure_filename(filename)) # type: ignore

def save(
self,
storage: FileStorage,
folder: Optional[str] = None,
name: Optional[str] = None
) -> str:
"""This saves the `storage` into this upload set.
A `storage` is a `werkzeug.datastructures.FileStorage`.
Expand All @@ -260,7 +305,8 @@ def save(self, storage, folder=None, name=None):

if folder is None and name is not None and "/" in name:
folder, name = os.path.split(name)

if storage.filename is None:
raise ValueError("Filename must not be empty!")
basename = self.get_basename(storage.filename)

if not self.file_allowed(storage, basename):
Expand Down Expand Up @@ -288,7 +334,7 @@ def save(self, storage, folder=None, name=None):
else:
return basename

def resolve_conflict(self, target_folder, basename):
def resolve_conflict(self, target_folder: str, basename: str) -> str:
"""
If a file with the selected name already exists in the target folder,
this method is called to resolve the conflict. It should return a new
Expand All @@ -314,7 +360,7 @@ def resolve_conflict(self, target_folder, basename):


@uploads_mod.route('/<setname>/<path:filename>')
def uploaded_file(setname, filename):
def uploaded_file(setname: UploadSet, filename: str) -> Any:
config = current_app.upload_set_config.get(setname)
if config is None:
abort(404)
Expand Down
41 changes: 29 additions & 12 deletions src/flask_uploads/test_helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,10 @@
- it has to be importable
- it can't be moved in the tests directory
"""
from typing import IO
from typing import Any
from typing import Optional

from werkzeug.datastructures import FileStorage


Expand All @@ -26,25 +30,38 @@ class TestingFileStorage(FileStorage):
:param headers: Multipart headers as a `werkzeug.Headers`. The default is
`None`.
"""
def __init__(self, stream=None, filename=None, name=None,
content_type='application/octet-stream', content_length=-1,
headers=None):
def __init__(
self,
stream: Optional[IO[bytes]] = None,
filename: Optional[str] = None,
name: Optional[str] = None,
content_type: str = 'application/octet-stream',
content_length: int = -1,
headers: Optional[Any] = None
) -> None:
FileStorage.__init__(
self, stream, filename, name=name,
content_type=content_type, content_length=content_length,
self,
stream,
filename,
name=name,
content_type=content_type,
content_length=content_length,
headers=None
)
self.saved = None
self.saved = None # type: Optional[str]

def save(self, dst, buffer_size=16384):
"""
This marks the file as saved by setting the `saved` attribute to the
name of the file it was saved to.
def save(self, dst: str, buffer_size: int = 16384) -> None: # type: ignore
"""This marks the file as saved.
The `saved` attribute gets set to the destination.
Although unused, `buffer_size` is required to stay compatible
with the signature of `werkzeug.datastructures.FileStorage.save`.
:param dst: The file to save to.
:param dst: The destination file name or path.
:param buffer_size: Ignored.
"""
if isinstance(dst, str):
self.saved = dst
else:
self.saved = dst.name
raise RuntimeError("dst currently has to be a `str`")
7 changes: 7 additions & 0 deletions tox.ini
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,13 @@ deps =
commands =
check-python-versions {posargs}

[testenv:mypy]
deps =
mypy
commands =
# do not lint tests yet; waiting for pytest 6.0 release
mypy --strict src {posargs}

[isort]
known_third_party = flask,flask_uploads,pytest,setuptools,sphinx_rtd_theme,werkzeug
force_single_line = True
Expand Down

0 comments on commit ba3ad70

Please sign in to comment.