Dev/v1/add admin module

include/ez/__init__.py

@@ -7,6 +7,7 @@
     from pathlib import Path
 
     from . import (
+        admin,
         data,
         database,
         events,
@@ -37,6 +38,7 @@
     "EZ_PYTHON_EXECUTABLE",
     "SITE_DIR",
 
+    "admin",
     "data",
     "database",
     "events",

include/utilities/utils.py

@@ -16,3 +16,14 @@ def wrapper(fn: Callable[B, Callable[P, T]]) -> Callable[P, T]:
         return fn(*args, **kwargs)
 
     return wrapper
+
+
+def spacify(text: str, sep: str = ' ') -> str:
+    """
+    Takes in a string of either PascalCase or camelCase and return it split to words
+    with the given separator. 
+
+    Default separator is a single space.
+    """
+    return ''.join(sep + char if char.isupper() else char.strip() for char in text).strip()
+

modules/admin/__api__/__init__.py

@@ -1,4 +1,6 @@
 from ..menu import MENU
 
+from . import view
+
 def add_item(name: str):
   MENU.add_item(name)

modules/admin/__api__/notifications.py

@@ -0,0 +1,33 @@
+from typing import Any, Iterable, overload
+
+from seamless.types import Renderable
+
+from ..user import User
+
+
+_EMPTY: Any = object()
+
+
+@overload
+def push(users: Iterable[User], message: str, /) -> None: ...
+@overload
+def push(users: Iterable[User], message: str, /, *, title: str) -> None: ...
+@overload
+def push(users: Iterable[User], custom: Renderable, /) -> None: ...
+
+def push(users: Iterable[User], message: str | Renderable, /, *, title: str = _EMPTY):
+    raise NotImplementedError
+
+
+@overload
+def push_all(message: str, /, *, exclude: Iterable[User] | None = None) -> None: ...
+@overload
+def push_all(message: str, /, *, title: str, exclude: Iterable[User] | None = None) -> None: ...
+@overload
+def push_all(custom: Renderable, /, *, exclude: Iterable[User] | None = None) -> None: ...
+
+def push_all(message: str | Renderable, /, *, title: str = _EMPTY, exclude: Iterable[User] | None = None):
+    if exclude is None:
+        exclude = []
+
+    raise NotImplementedError

modules/admin/__api__/view.py

@@ -0,0 +1 @@
+from ..views.settings_viewer import render_settings_viewer

modules/admin/__init__.py

@@ -0,0 +1,4 @@
+__deps__ = [
+    "plugins",
+    "web",
+]

modules/admin/__main__.py

@@ -0,0 +1 @@
+from . import router

modules/admin/authentication.py

@@ -0,0 +1,21 @@
+from uuid import uuid4
+from .user import User
+
+from . import context
+
+USERNAME = "admin"
+PASSWORD = "pwd"
+
+
+ADMIN_USER = User()
+
+
+def authenticate(username: str, password: str) -> context.Session | None:
+    if username != USERNAME or password != PASSWORD:
+        return None
+
+    user = ADMIN_USER
+
+    session = context.create_session(user, uuid4().hex)
+
+    return session

modules/admin/context.py

@@ -0,0 +1,261 @@
+"""
+The admin dashboard context module.
+
+The context for the dashboard is composed of multiple layers.
+
+First, the dashboard context is implemented by the ContextRoot class.
+
+The context root then contains a:
+- shared context (SharedContext)
+- user contexts (UserContext)
+- sessions (Session)
+
+The shared context is data that is shared between all logged in users.
+The user context is data for a specific user, but shared with all its sessions (open tabs, etc..)
+The session context is data for a specific session (connection, tab in browser, etc..)
+"""
+
+import ez
+
+from typing import TypeAlias, overload
+from utilities.utils import bind
+
+from .user import User
+
+
+SessionID: TypeAlias = str
+
+
+class SharedContext:
+    ...
+
+
+class SessionContext:
+    ...
+
+
+class UserContext:
+    def __init__(self, user: User) -> None:
+        self._user = user
+        self._sessions = []
+
+    @property
+    def user(self) -> User:
+        return self._user
+
+    @property
+    def sessions(self) -> list["Session"]:
+        return self._sessions.copy()
+
+    def create_session(self, session_id: str) -> "Session":
+        session = Session(self._user, session_id)
+        self._sessions.append(session)
+        return session
+
+    def connect_session(self, session: "Session"):
+        if session.user != self._user:
+            raise ValueError("Session user does not match user context.")
+        self._sessions.append(session)
+
+    def disconnect_session(self, session: "Session"):
+        try:
+            self._sessions.remove(session)
+        except ValueError:
+            return False
+        return True
+
+
+class Session:   
+    def __init__(self, user: User, session_id: SessionID) -> None:
+        self._id = session_id
+        self._user = user
+        self._context = SessionContext()
+
+    @property
+    def id(self) -> SessionID:
+        return self._id
+
+    @property
+    def user(self) -> User:
+        return self._user
+
+    @property
+    def context(self) -> SessionContext:
+        return self._context
+
+
+class ContextRoot:
+    _user_contexts: dict[User, UserContext]
+    _sessions: dict[SessionID, Session]
+
+    def __init__(self) -> None:
+        self._global_context = SharedContext()
+        self._user_contexts = {}
+        self._sessions = {}
+
+    @property
+    def global_context(self) -> SharedContext:
+        return self._global_context
+
+    def user_context(self, user: User) -> UserContext:
+        if user not in self._user_contexts:
+            self._user_contexts[user] = UserContext(user)
+        return self._user_contexts[user]
+
+    def get_user_context(self, user: User) -> UserContext | None:
+        try:
+            return self._user_contexts[user]
+        except KeyError:
+            return None
+
+    def _clear_user_context(self, user: User) -> None:
+        try:
+            del self._user_contexts[user]
+        except KeyError:
+            return
+
+    def session(self, session_id: SessionID) -> Session:
+        if session_id not in self._sessions:
+            raise KeyError(f"Session with ID '{session_id}' not found.")
+        return self._sessions[session_id]
+
+    def create_session(self, user: User, session_id: str) -> Session:
+        session = self.user_context(user).create_session(session_id)
+        self._sessions[session_id] = session
+        return session
+
+    def delete_session(self, session_id: SessionID) -> bool:
+        try:
+            session = self._sessions.pop(session_id)
+        except KeyError:
+            return False
+        context = self.get_user_context(session.user)
+
+        if context is None:
+            return False
+
+        result = context.disconnect_session(session)
+
+        if not context.sessions:
+            self._clear_user_context(session.user)
+
+        return result
+
+    def disconnect_user(self, user: User):
+        context = self.get_user_context(user)
+
+        if context is None:
+            return
+
+        for session in context.sessions:
+            self.delete_session(session.id)
+
+
+_SITE_CONTEXT = ContextRoot()
+
+
+@bind(_SITE_CONTEXT)
+def get_shared_context(ctx: ContextRoot):
+    def _get_global_context() -> SharedContext:
+        return ctx.global_context
+
+    return _get_global_context
+
+
+@bind(_SITE_CONTEXT)
+def get_context_root(ctx: ContextRoot):
+    def _get_site_context() -> ContextRoot:
+        return ctx
+
+    return _get_site_context
+
+
+@bind(_SITE_CONTEXT)
+def get_user_context(ctx: ContextRoot):
+    @overload
+    def _get_user_context(user: User, /) -> UserContext: ...
+    @overload
+    def _get_user_context(session: Session, /) -> UserContext: ...
+    @overload
+    def _get_user_context(session_id: SessionID, /) -> UserContext: ...
+
+    def _get_user_context(value: User | Session | SessionID, /) -> UserContext:
+        match value:
+            case User() as user:
+                result = ctx.get_user_context(user)
+                if result is None:
+                    raise ValueError("User context not found.")
+                return result
+            case Session() as session:
+                return _get_user_context(session.user)
+            case SessionID() as session_id:
+                return _get_user_context(ctx.session(session_id))
+            case _:
+                raise TypeError("Invalid argument type.")
+
+    return _get_user_context
+
+
+@bind(_SITE_CONTEXT)
+def create_session(ctx: ContextRoot):
+    def _create_session(user: User, session_id: SessionID) -> Session:
+        return ctx.create_session(user, session_id)
+
+    return _create_session
+
+
+@bind(_SITE_CONTEXT)
+def get_session(ctx: ContextRoot):
+    def _get_session(session_id: SessionID) -> Session:
+        return ctx.session(session_id)
+
+    return _get_session
+
+
+@bind(_SITE_CONTEXT)
+def close_session(ctx: ContextRoot):
+    @overload
+    def _close_session(session: Session, /) -> bool: ...
+    @overload
+    def _close_session(session_id: SessionID, /) -> bool: ...
+
+    def _close_session(session_id: Session | SessionID, /) -> bool:
+        if isinstance(session_id, Session):
+            session_id = session_id.id
+        return ctx.delete_session(session_id)
+
+    return _close_session
+
+
+@bind(_SITE_CONTEXT)
+def disconnect_user(ctx: ContextRoot):
+    def _disconnect_user(user: User):
+        ctx.disconnect_user(user)
+
+    return _disconnect_user
+
+
+@bind(_SITE_CONTEXT)
+def get_connected_users(ctx: ContextRoot):
+    def _get_users() -> list[User]:
+        return list(ctx._user_contexts.keys())
+
+    return _get_users
+
+
+def get_current_session() -> Session:
+    request = ez.web.http.request()
+    if request is None:
+        raise ValueError("No request found")
+    try:
+        sid = request.session["ez-admin:sid"]
+    except KeyError:
+        raise ValueError("Missing session id")
+    return get_session(sid)
+
+
+def get_current_user():
+    return get_current_session().user
+
+
+del _SITE_CONTEXT

modules/admin/requests.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class LoginRequest(BaseModel):
+    username: str
+    password: str