[WIP] Flask3 migration, general cleanup

portal/__init__.py

@@ -1,13 +1,14 @@
+import sys
+import logging.handlers
 from flask import Flask
 from flask_wtf.csrf import CSRFProtect
 from datetime import timedelta
 
 # from flask import Markup
 from flask_misaka import Misaka
-import logging.handlers
-import sys
 
 from portal import app_logging
+
 logger = app_logging.init_logger()
 
 __author__ = "MANIAC Lab <[email protected]>"
@@ -24,14 +25,17 @@
         app.config.from_pyfile(config_file)
         logger.info("Read config file from sys.argv[1]")
     except:
-        logger.error("Could not read config location from {}".format(sys.argv[1]))
+        logger.error(f"Could not read config location from {sys.argv[1]}")
 else:
     app.config.from_pyfile("portal.conf")
     logger.info("Read config file from portal.conf")
 
-from portal import k8s_api
-k8s_api.load_kube_config()
-k8s_api.start_notebook_manager()
+print(app.config)
+if app.config["K8S_ENABLED"]:
+    from portal import k8s_api
+
+    k8s_api.load_kube_config()
+    k8s_api.start_notebook_manager()
 
 app.url_map.strict_slashes = False
 app.permanent_session_lifetime = timedelta(minutes=1440)

portal/admin.py

@@ -1,121 +1,161 @@
-import requests
 import json
-from matplotlib.figure import Figure
-from io import BytesIO
 import base64
-from flask import session
+from io import BytesIO
+from datetime import datetime
+import requests
 from dateutil.parser import parse
-from datetime import datetime 
-from datetime import timedelta
+from flask import session
+from matplotlib.figure import Figure
 from portal import logger
 from portal import app
 
 ciconnect_api_token = app.config["CONNECT_API_TOKEN"]
 ciconnect_api_endpoint = app.config["CONNECT_API_ENDPOINT"]
 mailgun_api_token = app.config["MAILGUN_API_TOKEN"]
-params = {'token': ciconnect_api_token}
+params = {"token": ciconnect_api_token}
+MAX_TIMEOUT = 10  # seconds
+
 
 def authorized():
-    return session.get('admin') == 'admin'
+    return session.get("admin") == "admin"
+
 
 def get_usernames(group):
     if authorized():
         try:
-            url = ciconnect_api_endpoint + "/v1alpha1/groups/" + group + "/members?token=" + ciconnect_api_token
-            users = requests.get(url).json()
-            return [member['user_name']for member in users['memberships']]
-        except:
-            logger.error('Error getting usernames')
+            url = (
+                ciconnect_api_endpoint
+                + "/v1alpha1/groups/"
+                + group
+                + "/members?token="
+                + ciconnect_api_token
+            )
+            users = requests.get(url, timeout=MAX_TIMEOUT).json()
+            return [member["user_name"] for member in users["memberships"]]
+        except requests.exceptions.RequestException as e:
+            logger.error("Error getting usernames: %s", e)
+
 
 def get_user_profiles(group):
+    profiles = []
     if authorized():
         try:
             usernames = get_usernames(group)
 
             multiplex_json = {}
             for username in usernames:
-                multiplex_json['/v1alpha1/users/' + username + '?token=' + ciconnect_api_token] = {'method': 'GET'}
+                multiplex_json[
+                    "/v1alpha1/users/" + username + "?token=" + ciconnect_api_token
+                ] = {"method": "GET"}
 
-            url = ciconnect_api_endpoint + '/v1alpha1/multiplex'
-            resp = requests.post(url, params=params, json=multiplex_json)
+            url = ciconnect_api_endpoint + "/v1alpha1/multiplex"
+            resp = requests.post(
+                url, params=params, json=multiplex_json, timeout=MAX_TIMEOUT
+            )
             data = resp.json()
 
-            profiles = []
             for entry in data:
-                user = json.loads(data[entry]['body'])
-                username = user['metadata']['unix_name']
-                email = user['metadata']['email']
-                join_date = parse(user['metadata']['join_date']).strftime('%Y-%m-%d')
-                institution = user['metadata']['institution']
-                name = user['metadata']['name']
-                profiles.append({'username': username, 'email': email, 'join_date': join_date, 'institution': institution, 'name': name})
-            return profiles
-        except:
-            logger.error('Error getting user profiles')
+                user = json.loads(data[entry]["body"])
+                username = user["metadata"]["unix_name"]
+                email = user["metadata"]["email"]
+                join_date = parse(user["metadata"]["join_date"]).strftime("%Y-%m-%d")
+                institution = user["metadata"]["institution"]
+                name = user["metadata"]["name"]
+                profiles.append(
+                    {
+                        "username": username,
+                        "email": email,
+                        "join_date": join_date,
+                        "institution": institution,
+                        "name": name,
+                    }
+                )
+        except requests.exceptions.RequestException as e:
+            logger.error("Error getting user profiles: %s", e)
+    return profiles
+
 
 def get_email_list(group):
+    email_list = []
     if authorized():
-        email_list = []
         profiles = get_user_profiles(group)
         for profile in profiles:
-            email_list.append(profile['email'])
-        return email_list
+            email_list.append(profile["email"])
+    return email_list
+
 
 def update_user_institution(username, institution):
     if authorized():
         try:
-            json_data = {'apiVersion': 'v1alpha1', 'kind': 'User', 'metadata': {'institution': institution}}
+            json_data = {
+                "apiVersion": "v1alpha1",
+                "kind": "User",
+                "metadata": {"institution": institution},
+            }
             url = ciconnect_api_endpoint + "/v1alpha1/users/" + username
-            resp = requests.put(url, params=params, json=json_data)
-            logger.info("Updated user %s. Set institution to %s." %(username, institution))
+            resp = requests.put(url, params=params, json=json_data, timeout=MAX_TIMEOUT)
+            logger.info("Updated user %s. Set institution to %s", username, institution)
             return resp
-        except Exception as err:
-            logger.error("Error updating institution for user %s." %username)
-
+        except requests.exceptions.RequestException as e:
+            logger.error("Error updating institution for user %s: %s", username, e)
+
+
 def email_users(sender, recipients, subject, body):
     if authorized():
-        try: 
+        try:
             logger.info("Sending email...")
-            resp = requests.post("https://api.mailgun.net/v3/api.ci-connect.net/messages",
+            resp = requests.post(
+                "https://api.mailgun.net/v3/api.ci-connect.net/messages",
                 auth=("api", mailgun_api_token),
                 data={
                     "from": "<" + sender + ">",
                     "to": [sender],
                     "bcc": recipients,
                     "subject": subject,
-                    "text": body
-                }
+                    "text": body,
+                },
+                timeout=MAX_TIMEOUT,
             )
             return resp
-        except:
-            logger.error("Error sending email to all users")
+        except requests.exceptions.RequestException as e:
+            logger.error("Error sending email to all users: %s", e)
+
 
 def plot_users_by_join_date(users):
     if authorized():
         try:
-            xvalues_format = "%m-%Y" 
+            xvalues_format = "%m-%Y"
             xvalues_set = set()
             for user in users:
-                join_date = datetime.strptime(user['join_date'], '%Y-%m-%d')
-                user['jd'] = join_date
-                xvalue = datetime(join_date.year, join_date.month, 1).strftime(xvalues_format)
+                join_date = datetime.strptime(user["join_date"], "%Y-%m-%d")
+                user["jd"] = join_date
+                xvalue = datetime(join_date.year, join_date.month, 1).strftime(
+                    xvalues_format
+                )
                 xvalues_set.add(xvalue)
             xvalues = list(xvalues_set)
-            xvalues.sort(key=lambda x:datetime.strptime(x, xvalues_format))
+            xvalues.sort(key=lambda x: datetime.strptime(x, xvalues_format))
             yvalues = [0] * len(xvalues)
             for i in range(len(xvalues)):
                 xvalue = datetime.strptime(xvalues[i], xvalues_format)
-                L = list(filter(lambda u : ((xvalue.year - u['jd'].year) * 12) + (xvalue.month - u['jd'].month) >= 0, users))
-                yvalues[i] = len(L)
+                l = list(
+                    filter(
+                        lambda u: ((xvalue.year - u["jd"].year) * 12)
+                        + (xvalue.month - u["jd"].month)
+                        >= 0,
+                        users,
+                    )
+                )
+                yvalues[i] = len(l)
             fig = Figure(figsize=(16, 8), dpi=80, tight_layout=True)
             ax = fig.subplots()
             ax.plot(xvalues, yvalues)
-            ax.set_xlabel('Month')
-            ax.set_ylabel('Number of users')
-            ax.set_title('Number of users by month')
+            ax.set_xlabel("Month")
+            ax.set_ylabel("Number of users")
+            ax.set_title("Number of users by month")
             buf = BytesIO()
-            fig.savefig(buf, format='png')
+            fig.savefig(buf, format="png")
             data = base64.b64encode(buf.getbuffer()).decode("ascii")
             return data
         except:
-            logger.error('Error generating user by join date plot')
+            logger.error("Error generating user by join date plot")

portal/app_logging.py

@@ -1,19 +1,23 @@
 import logging
 
+
 def init_logger():
     logger = logging.getLogger()
     logger.setLevel(logging.INFO)
     ch = logging.StreamHandler()
     ch.setLevel(logging.INFO)
-    fh = logging.FileHandler('ciconnect-portal.log')
+    fh = logging.FileHandler("ciconnect-portal.log")
     fh.setLevel(logging.INFO)
-    formatter = logging.Formatter("%(asctime)s;%(module)s;%(funcName)s;%(levelname)s;%(message)s")
+    formatter = logging.Formatter(
+        "%(asctime)s;%(module)s;%(funcName)s;%(levelname)s;%(message)s"
+    )
     ch.setFormatter(formatter)
     fh.setFormatter(formatter)
     logger.propagate = False
     logger.addHandler(ch)
     logger.addHandler(fh)
     return logger
 
+
 def get_logger():
-    return logging.getLogger()
+    return logging.getLogger()