refactor login handling: replace inline logic with perform_login function for improved clarity and maintainability

nice.py

@@ -36,7 +36,7 @@ from starlette.middleware.base import BaseHTTPMiddleware
 import typer
 import logging
 
-import traceback
+# traceback was previously imported for debugging but is unused
 
 # TODO fix ssl
 VERIFY_SSL = False
@@ -449,39 +449,26 @@ def upload_files(q, rqst):
     )
 
 
+def perform_login(user: str, pw: str) -> dict:
+    """Module-level blocking login function suitable for run.cpu_bound.
 
-
+    Returns a dict with keys:
-
+      - error: str on error
-@ui.page("/login")
+      - session: requests.Session on success
-def login() -> Optional[RedirectResponse]:
+      - token: csrf token if found
-    dark = ui.dark_mode()
+      - success: bool
-    dark.enable()
+    """
-
+    s = requests.session()
-    def try_login() -> (
-        None
-    ):  # local function to avoid passing username and password as arguments
-        # Because I haven't gotten around to implementing a proper login system yet
-        # we just log in via requests and store teh session in the global rqst variable
-        user = username.value
-        pw = password.value
-
-        global rqst, LOGIN_SUCCESS, LOGGED_IN_USER
-        rqst = requests.session()
     try:
-            rsp = rqst.get(LOGIN_URL, verify=VERIFY_SSL)
+        r = s.get(LOGIN_URL, verify=VERIFY_SSL, timeout=10)
-        except requests.exceptions.ConnectionError as e:
+    except Exception as e:
-            ui.notify("Connection error!", color="negative")
         logger.debug(e)
-            print(traceback.format_exc())
+        return {"error": str(e)}
-            return
 
-        token = (
+    try:
-            BeautifulSoup(rsp.content, "html.parser").find("input").attrs["value"]
+        token = BeautifulSoup(r.content, "html.parser").find("input").attrs.get("value")
-        )  # , attr={"name": "csrfmiddlewaretoken"}).attrs("value")
+    except Exception:
-
+        token = None
-        # token = rsp.cookies["csrftoken"]
-        # header = {"X-CSRFToken": token}
-        # cookies = {"csrftoken": token}
 
     data = {
         "username": user,
@@ -490,35 +477,53 @@ def login() -> Optional[RedirectResponse]:
         "next": "/",
     }
 
-        rqst.headers["Referer"] = LOGIN_URL
+    s.headers["Referer"] = LOGIN_URL
 
-        rsp = rqst.post(
+    try:
-            LOGIN_URL,
+        r2 = s.post(LOGIN_URL, data=data, verify=VERIFY_SSL, timeout=10)
-            data=data,
+    except Exception as e:
-            # headers=header,
+        logger.debug(e)
-            # cookies=cookies
+        return {"error": str(e)}
-            verify=VERIFY_SSL
-        )
-
-        print(rsp.content)
-
-        soup = BeautifulSoup(rsp.content, "html.parser")
 
+    soup = BeautifulSoup(r2.content, "html.parser")
     if soup.find("button") and soup.find("button").find(string="Login"):
-            print("login fail")
+        return {"session": s, "success": False}
-            pass
     else:
-            print("login success")
+        return {"session": s, "success": True, "token": token}
+
+
+
+
+
+@ui.page("/login")
+def login() -> Optional[RedirectResponse]:
+    dark = ui.dark_mode()
+    dark.enable()
+
+    async def try_login() -> None:
+        # Run blocking network calls off the UI thread to keep the page responsive.
+        user = username.value
+        pw = password.value
+
+        global rqst, LOGIN_SUCCESS, LOGGED_IN_USER
+
+        # Use the module-level perform_login function (picklable) to avoid
+        # multiprocessing pickling errors when using run.cpu_bound.
+        result = await run.cpu_bound(perform_login, user, pw)
+
+        if result.get("error"):
+            ui.notify("Connection error!", color="negative")
+            logger.debug(result.get("error"))
+            return
+
+        if result.get("success"):
+            rqst = result.get("session")
+            token = result.get("token")
+            if token:
+                rqst.headers["X-CSRFToken"] = token
             LOGIN_SUCCESS = True
             LOGGED_IN_USER = user
             user_info_ui.refresh()
-            # user_label.clear()
-            # user_label.text = f"User: {user}"
-        # print(rqst.headers)
-
-        rqst.headers["X-CSRFToken"] = token
-
-        if LOGIN_SUCCESS:
             ui.notify("Logged in!", color="positive")
             ui.navigate.to("/")
         else:
@@ -561,7 +566,7 @@ def main_page():
         globals(), "LOGIN_SUCCESS", backward=lambda v: not v
     )
 
-    updates = ui.list()
+    # placeholder for update list (previously assigned but unused)
 
     loaded_series_ui()