feature: Allow to decorate upload views

For authentication or other needs.

For example, when you need to ensure that user allowed to upload files,
you might need to provide decorator to check that on `setup_tus` call.

CHANGELOG.rst

@@ -1,6 +1,8 @@
 1.0.0a1 (In Development)
 ========================
 
+- Allow to decorate upload views for authentication or other (for example *to check
+  whether entity for upload exists or not*) needs
 - Allow to upload on named upload paths, when using named upload URLs
 - Ensure named upload URLs (e.g. ``/user/{username}/uploads``) works as well
 - Ensure package is typed by adding ``py.typed``

aiohttp_tus/annotations.py

@@ -1,5 +1,9 @@
-from typing import Any, Dict, Mapping
+from typing import Any, Callable, Dict, Mapping
 
+from aiohttp.web_middlewares import _Handler as Handler
+
+
+Decorator = Callable[[Handler], Handler]
 
 DictStrAny = Dict[str, Any]
 DictStrBytes = Dict[str, bytes]

aiohttp_tus/tus.py

@@ -3,6 +3,7 @@
 from aiohttp import web
 
 from . import views
+from .annotations import Decorator, Handler
 from .constants import APP_TUS_CONFIG_KEY, ROUTE_RESOURCE, ROUTE_UPLOAD
 from .data import TusConfig
 
@@ -13,21 +14,31 @@ def setup_tus(
     upload_path: Path,
     upload_url: str = "/uploads",
     allow_overwrite_files: bool = False,
+    decorator: Decorator = None,
 ) -> web.Application:
+    """Setup tus protocol server implementation for aiohttp.web application."""
+
+    def decorate(handler: Handler) -> Handler:
+        if decorator is None:
+            return handler
+        return decorator(handler)
+
     # Store tus config in application
     app[APP_TUS_CONFIG_KEY] = TusConfig(
         upload_path=upload_path, allow_overwrite_files=allow_overwrite_files,
     )
 
     # Views for upload management
     app.router.add_options(upload_url, views.upload_options, name=ROUTE_UPLOAD)
-    app.router.add_get(upload_url, views.upload_details)
-    app.router.add_post(upload_url, views.start_upload)
+    app.router.add_get(upload_url, decorate(views.upload_details))
+    app.router.add_post(upload_url, decorate(views.start_upload))
 
     # Views for resource management
     resource_url = "/".join((upload_url.rstrip("/"), r"{resource_uid}"))
-    app.router.add_head(resource_url, views.resource_details, name=ROUTE_RESOURCE)
-    app.router.add_delete(resource_url, views.delete_resource)
-    app.router.add_patch(resource_url, views.upload_resource)
+    app.router.add_head(
+        resource_url, decorate(views.resource_details), name=ROUTE_RESOURCE
+    )
+    app.router.add_delete(resource_url, decorate(views.delete_resource))
+    app.router.add_patch(resource_url, decorate(views.upload_resource))
 
     return app

tests/test_tus.py

@@ -1,4 +1,5 @@
 import tempfile
+from functools import partial
 from pathlib import Path
 
 try:
@@ -8,33 +9,83 @@
 
 import pytest
 import tus
-from aiohttp import web
+from aiohttp import hdrs, web
 from aiohttp.test_utils import TestClient
 
 from aiohttp_tus import setup_tus
+from aiohttp_tus.annotations import Decorator, Handler
 from aiohttp_tus.constants import APP_TUS_CONFIG_KEY
 from aiohttp_tus.data import TusConfig
 
 
+SECRET_TOKEN = "secret-token"
 TEST_DATA_PATH = Path(__file__).parent / "test-data"
 
 
 @pytest.fixture
 def aiohttp_test_client(aiohttp_client):
     @asynccontextmanager
-    async def factory(*, upload_url: str, upload_suffix: str = None) -> TestClient:
+    async def factory(
+        *, upload_url: str, upload_suffix: str = None, decorator: Decorator = None
+    ) -> TestClient:
         with tempfile.TemporaryDirectory(prefix="aiohttp_tus") as temp_path:
             base_path = Path(temp_path)
             app = setup_tus(
                 web.Application(),
                 upload_path=base_path / upload_suffix if upload_suffix else base_path,
                 upload_url=upload_url,
+                decorator=decorator,
             )
             yield await aiohttp_client(app)
 
     return factory
 
 
+def get_upload_url(client: TestClient, upload_url: str) -> str:
+    return f"http://{client.host}:{client.port}{upload_url}"
+
+
+def login_required(handler: Handler) -> Handler:
+    async def decorator(request: web.Request) -> web.StreamResponse:
+        header = request.headers.get(hdrs.AUTHORIZATION)
+        if header is None or header != f"Token {SECRET_TOKEN}":
+            raise web.HTTPForbidden()
+        return await handler(request)
+
+    return decorator
+
+
+async def test_decorated_upload_200(aiohttp_test_client, loop):
+    async with aiohttp_test_client(
+        upload_url="/uploads", decorator=login_required
+    ) as client:
+        upload = partial(
+            tus.upload,
+            file_name="hello.txt",
+            headers={"Authorization": "Token secret-token"},
+        )
+        with open(TEST_DATA_PATH / "hello.txt", "rb") as handler:
+            await loop.run_in_executor(
+                None, upload, handler, get_upload_url(client, "/uploads")
+            )
+
+
+async def test_decorated_upload_403(aiohttp_test_client, loop):
+    async with aiohttp_test_client(
+        upload_url="/uploads", decorator=login_required
+    ) as client:
+        upload = partial(
+            tus.upload,
+            file_name="hello.txt",
+            headers={"Authorization": "Token not-secret-token"},
+        )
+        with open(TEST_DATA_PATH / "hello.txt", "rb") as handler:
+            with pytest.raises(tus.TusError):
+                await loop.run_in_executor(
+                    None, upload, handler, get_upload_url(client, "/uploads")
+                )
+
+
 @pytest.mark.parametrize(
     "upload_url, upload_suffix, tus_upload_url, match_info",
     (
@@ -54,11 +105,13 @@ async def test_upload(
     async with aiohttp_test_client(
         upload_url=upload_url, upload_suffix=upload_suffix
     ) as client:
-        upload_url = f"http://{client.host}:{client.port}{tus_upload_url}"
         test_upload_path = TEST_DATA_PATH / "hello.txt"
 
+        upload = partial(tus.upload, file_name="hello.txt")
         with open(test_upload_path, "rb") as handler:
-            await loop.run_in_executor(None, tus.upload, handler, upload_url)
+            await loop.run_in_executor(
+                None, upload, handler, get_upload_url(client, tus_upload_url)
+            )
 
         config: TusConfig = client.app[APP_TUS_CONFIG_KEY]
         expected_upload_path = config.resolve_upload_path(match_info) / "hello.txt"