UPdate.

src/snek/system/service.py

@@ -39,14 +39,18 @@ class BaseService:
         for record in self.app.db.query(sql, *args):
             yield record
 
-    async def get(self, uid=None, **kwargs):
-        kwargs["deleted_at"] = None 
-        if uid:
+    async def get(self, *args, **kwargs):
+        if not "deleted_at" in kwargs:
+            kwargs["deleted_at"] = None 
+        uid = kwargs.get("uid")
+        if args:
+            uid = args[0]
+        if uid or "uid" in kwargs:
             result = await self.cache.get(uid)
             if result and result.__class__ == self.mapper.model_class:
                 return result
             kwargs["uid"] = uid
-
+        
         result = await self.mapper.get(**kwargs)
         if result:
             await self.cache.set(result["uid"], result)

src/snek/view/channel.py

@@ -16,6 +16,9 @@ register_heif_opener()
 from snek.view.drive import DriveApiView 
 
 class ChannelDriveApiView(DriveApiView):
+    
+    login_required = True
+
     async def get_target(self):
         target = await self.services.channel.get_home_folder(self.request.match_info.get("channel_uid"))
         target.mkdir(parents=True, exist_ok=True)
@@ -25,6 +28,9 @@ class ChannelDriveApiView(DriveApiView):
         return f"/channel/{self.request.match_info.get('channel_uid')}/drive/{urllib.parse.quote(rel)}"
 
 class ChannelAttachmentView(BaseView):
+
+    login_required=True
+
     async def get(self):
         relative_path = self.request.match_info.get("relative_url")
         channel_attachment = await self.services.channel_attachment.get(
@@ -158,7 +164,8 @@ class ChannelAttachmentView(BaseView):
 
 
 class ChannelAttachmentUploadView(BaseView):
-
+    
+    login_required = True
 
     async def get(self):
    
@@ -214,6 +221,9 @@ class ChannelAttachmentUploadView(BaseView):
 
 
 class ChannelView(BaseView):
+
+    login_required = True
+
     async def get(self):
         channel_name = self.request.match_info.get("channel")
         if channel_name is None:

src/snek/view/container.py

@@ -4,6 +4,7 @@ from aiohttp import web
 
 class ContainerView(BaseView):
 
+    login_required= True
 
     async def stdout_event_handler(self, ws, data):
         try:

src/snek/view/drive.py

@@ -12,6 +12,8 @@ from snek.system.view import BaseView
 
 class DriveView(BaseView):
 
+    login_required = True
+
     async def get(self):
         target = await self.services.user.get_home_folder(self.session.get("uid"))
         rel_path = self.request.match_info.get("rel_path", "")
@@ -104,176 +106,3 @@ class DriveApiView(BaseView):
                 "url": str(url),
             }
         )
-
-
-class DriveView222(BaseView):
-    PAGE_SIZE = 20
-
-    async def base_path(self):
-        return await self.services.user.get_home_folder(self.session.get("uid"))
-
-    async def get_full_path(self, rel_path):
-        base_path = await self.base_path()
-        safe_path = os.path.normpath(unquote(rel_path or ""))
-        full_path = os.path.abspath(os.path.join(base_path, safe_path))
-        if not full_path.startswith(os.path.abspath(base_path)):
-            raise web.HTTPForbidden(reason="Invalid path")
-        return full_path
-
-    async def make_absolute_url(self, rel_path):
-        rel_path = rel_path.lstrip("/")
-        url = str(self.request.url.with_path(f"/drive/{quote(rel_path)}"))
-        return url
-
-    async def entry_details(self, dir_path, entry, parent_rel_path):
-        entry_path = os.path.join(dir_path, entry)
-        stat = os.stat(entry_path)
-        is_dir = os.path.isdir(entry_path)
-        mimetype = (
-            None
-            if is_dir
-            else (mimetypes.guess_type(entry_path)[0] or "application/octet-stream")
-        )
-        size = stat.st_size if not is_dir else None
-        created_at = datetime.fromtimestamp(stat.st_ctime).isoformat()
-        updated_at = datetime.fromtimestamp(stat.st_mtime).isoformat()
-        rel_entry_path = os.path.join(parent_rel_path, entry).replace("\\", "/")
-        return {
-            "name": entry,
-            "type": "dir" if is_dir else "file",
-            "mimetype": mimetype,
-            "size": size,
-            "created_at": created_at,
-            "updated_at": updated_at,
-            "absolute_url": await self.make_absolute_url(rel_entry_path),
-        }
-
-    async def get(self):
-        rel_path = self.request.match_info.get("rel_path", "")
-        full_path = await self.get_full_path(rel_path)
-        page = int(self.request.query.get("page", 1))
-        page_size = int(self.request.query.get("page_size", self.PAGE_SIZE))
-        abs_url = await self.make_absolute_url(rel_path)
-
-        if not os.path.exists(full_path):
-            raise web.HTTPNotFound(reason="Path not found")
-
-        if os.path.isdir(full_path):
-            entries = os.listdir(full_path)
-            entries.sort()
-            start = (page - 1) * page_size
-            end = start + page_size
-            paged_entries = entries[start:end]
-            details = [
-                await self.entry_details(full_path, entry, rel_path)
-                for entry in paged_entries
-            ]
-            return web.json_response(
-                {
-                    "path": rel_path,
-                    "absolute_url": abs_url,
-                    "entries": details,
-                    "total": len(entries),
-                    "page": page,
-                    "page_size": page_size,
-                }
-            )
-        else:
-            with open(full_path, "rb") as f:
-                content = f.read()
-            mimetype = mimetypes.guess_type(full_path)[0] or "application/octet-stream"
-            headers = {"X-Absolute-Url": abs_url}
-            return web.Response(body=content, content_type=mimetype, headers=headers)
-
-    async def post(self):
-        rel_path = self.request.match_info.get("rel_path", "")
-        full_path = await self.get_full_path(rel_path)
-        abs_url = await self.make_absolute_url(rel_path)
-        if os.path.exists(full_path):
-            raise web.HTTPConflict(reason="File or directory already exists")
-        data = await self.request.post()
-        if data.get("type") == "dir":
-            os.makedirs(full_path)
-            return web.json_response(
-                {"status": "created", "type": "dir", "absolute_url": abs_url}
-            )
-        else:
-            file_field = data.get("file")
-            if not file_field:
-                raise web.HTTPBadRequest(reason="No file uploaded")
-            with open(full_path, "wb") as f:
-                f.write(file_field.file.read())
-            return web.json_response(
-                {"status": "created", "type": "file", "absolute_url": abs_url}
-            )
-
-    async def put(self):
-        rel_path = self.request.match_info.get("rel_path", "")
-        full_path = await self.get_full_path(rel_path)
-        abs_url = await self.make_absolute_url(rel_path)
-        if not os.path.exists(full_path):
-            raise web.HTTPNotFound(reason="File not found")
-        if os.path.isdir(full_path):
-            raise web.HTTPBadRequest(reason="Cannot overwrite directory")
-        body = await self.request.read()
-        with open(full_path, "wb") as f:
-            f.write(body)
-        return web.json_response({"status": "updated", "absolute_url": abs_url})
-
-    async def delete(self):
-        rel_path = self.request.match_info.get("rel_path", "")
-        full_path = await self.get_full_path(rel_path)
-        abs_url = await self.make_absolute_url(rel_path)
-        if not os.path.exists(full_path):
-            raise web.HTTPNotFound(reason="Path not found")
-        if os.path.isdir(full_path):
-            os.rmdir(full_path)
-            return web.json_response(
-                {"status": "deleted", "type": "dir", "absolute_url": abs_url}
-            )
-        else:
-            os.remove(full_path)
-            return web.json_response(
-                {"status": "deleted", "type": "file", "absolute_url": abs_url}
-            )
-
-
-class DriveViewi2(BaseView):
-
-    login_required = True
-
-    async def get(self):
-
-        drive_uid = self.request.match_info.get("drive")
-
-        before = self.request.query.get("before")
-        filters = {}
-        if before:
-            filters["created_at__lt"] = before
-
-        if drive_uid:
-            filters["drive_uid"] = drive_uid
-            drive = await self.services.drive.get(uid=drive_uid)
-            drive_items = []
-
-            async for item in self.services.drive_item.find(**filters):
-                record = item.record
-                record["url"] = "/drive.bin/" + record["uid"] + "." + item.extension
-                drive_items.append(record)
-            return web.json_response(drive_items)
-
-        user = await self.services.user.get(uid=self.session.get("uid"))
-
-        drives = []
-        async for drive in self.services.drive.get_by_user(user["uid"]):
-            record = drive.record
-            record["items"] = []
-            async for item in drive.items:
-                drive_item_record = item.record
-                drive_item_record["url"] = (
-                    "/drive.bin/" + drive_item_record["uid"] + "." + item.extension
-                )
-                record["items"].append(item.record)
-            drives.append(record)
-
-        return web.json_response(drives)

src/snek/view/register_form.py

@@ -30,6 +30,9 @@ from snek.system.view import BaseFormView
 
 
 class RegisterFormView(BaseFormView):
+    
+    login_required = False
+
     form = RegisterForm
 
     async def submit(self, form):

src/snek/view/repository.py

@@ -11,6 +11,9 @@ from snek.system.view import BaseView
 
 
 class BareRepoNavigator:
+
+    login_required = True
+
     def __init__(self, repo_path):
         """Initialize the navigator with a bare repository path."""
         try:

src/snek/view/stats.py

@@ -7,6 +7,8 @@ from snek.system.view import BaseView
 
 class StatsView(BaseView):
 
+    login_required = True
+
     async def get(self):
         data = await self.app.cache.get_stats()
         data = json.dumps({"total": len(data), "stats": data}, default=str, indent=1)

src/snek/view/status.py

@@ -27,6 +27,9 @@ from snek.system.view import BaseView
 
 
 class StatusView(BaseView):
+
+    login_required = True
+
     async def get(self):
         memberships = []
         user = {}

src/snek/view/threads.py

@@ -3,6 +3,8 @@ from snek.system.view import BaseView
 
 class ThreadsView(BaseView):
 
+    login_required = True
+
     async def get(self):
         threads = []
         user = await self.services.user.get(uid=self.session.get("uid"))

src/snek/view/upload.py

@@ -21,6 +21,8 @@ from snek.system.view import BaseView
 
 class UploadView(BaseView):
 
+    login_required = True
+
     async def get(self):
         uid = self.request.match_info.get("uid")
         drive_item = await self.services.drive_item.get(uid)

src/snek/view/user.py

@@ -3,6 +3,8 @@ from snek.system.view import BaseView
 
 class UserView(BaseView):
 
+    login_required = True
+
     async def get(self):
         user_uid = self.request.match_info.get("user")
         user = await self.services.user.get(uid=user_uid)