trash

Author: pcrespov

services/web/server/src/simcore_service_webserver/login/decorators.py

@@ -1,7 +1,9 @@
 import functools
 import inspect
+from typing import cast
 
 from aiohttp import web
+from models_library.users import UserID
 from servicelib.aiohttp.typing_extension import HandlerAnyReturn
 from servicelib.request_keys import RQT_USERID_KEY
 
@@ -69,3 +71,7 @@ async def _wrapper(request: web.Request):
         return await handler(request)
 
     return _wrapper
+
+
+def get_user_id(request: web.Request) -> UserID:
+    return cast(UserID, request[RQT_USERID_KEY])

services/web/server/src/simcore_service_webserver/projects/_trash_api.py

@@ -0,0 +1,7 @@
+from aiohttp import web
+from models_library.products import ProductName
+from models_library.users import UserID
+
+
+async def empty_trash(app: web.Application, product_name: ProductName, user_id: UserID):
+    raise NotImplementedError

services/web/server/src/simcore_service_webserver/trash/__init__.py

@@ -0,0 +1,46 @@
+import functools
+
+from aiohttp import web
+from servicelib.aiohttp import status
+from servicelib.aiohttp.typing_extension import Handler
+from simcore_postgres_database.utils_tags import (
+    TagNotFoundError,
+    TagOperationNotAllowedError,
+)
+from simcore_service_webserver.products.api import get_product_name
+
+from .._meta import API_VTAG as VTAG
+from ..login.decorators import get_user_id, login_required
+from ..security.decorators import permission_required
+from . import _api
+
+
+def _handle_trash_exceptions(handler: Handler):
+    @functools.wraps(handler)
+    async def wrapper(request: web.Request) -> web.StreamResponse:
+        try:
+            return await handler(request)
+
+        except TagNotFoundError as exc:
+            raise web.HTTPNotFound(reason=f"{exc}") from exc
+
+        except TagOperationNotAllowedError as exc:
+            raise web.HTTPForbidden(reason=f"{exc}") from exc
+
+    return wrapper
+
+
+routes = web.RouteTableDef()
+
+
+@routes.delete(f"/{VTAG}/trash", name="empty_trash")
+@login_required
+@permission_required("project.delete")
+@_handle_trash_exceptions
+async def empty_trash(request: web.Request):
+    user_id = get_user_id(request)
+    product_name = get_product_name(request)
+
+    await _api.empty_trash(request.app, product_name=product_name, user_id=user_id)
+
+    return web.json_response(status=status.HTTP_204_NO_CONTENT)