Update to check for config theme and default to default when not pres…

…ent.

self-registration/app/main.py

@@ -123,15 +123,19 @@ def assign_user_to_group(user, group_name):
     return True
 
 
+def get_theme():
+    return config.get("theme", DEFAULT_THEME)
+
+
 def get_template_context(request: Request, error_message: str = None):
     if (error_message is None) or (error_message == ""):
-        return {"url_prefix": url_prefix, "request": request, **DEFAULT_THEME}
+        return {"url_prefix": url_prefix, "request": request, **get_theme()}
     else:
         return {
             "url_prefix": url_prefix,
             "request": request,
             "error_message": error_message,
-            **DEFAULT_THEME,
+            **get_theme(),
         }
 
 
@@ -167,7 +171,7 @@ async def validate_submission(request: Request, email: str = Form(...), coupon_c
                             "temporary_password": temporary_password,
                             "user_id": user["id"],
                             "expiration_date": expiration_date.strftime("%m-%d-%Y"),
-                            **DEFAULT_THEME,
+                            **get_theme(),
                         },
                     )
                 else:

src/nebari_plugin_self_registration/__init__.py

@@ -1,49 +1,50 @@
 import inspect
 import sys
 import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Union
 
-from nebari.schema import Base
 from _nebari.stages.base import NebariTerraformStage
+from _nebari.stages.tf_objects import NebariKubernetesProvider, NebariTerraformState
 from nebari.hookspecs import NebariStage, hookimpl
-from pathlib import Path
-from typing import Any, Dict, List, Optional, Union
-from _nebari.stages.tf_objects import (
-    NebariKubernetesProvider,
-    NebariTerraformState,
-)
+from nebari.schema import Base
 
 NUM_ATTEMPTS = 10
 TIMEOUT = 10
 
 CLIENT_NAME = "self-registration"
 
+
 class SelfRegistrationAffinitySelectorConfig(Base):
     default: str
     app: Optional[str] = ""
     job: Optional[str] = ""
 
+
 class SelfRegistrationConfig(Base):
     enabled: Optional[bool] = True
     selector: Union[SelfRegistrationAffinitySelectorConfig, str] = "general"
 
+
 class SelfRegistrationConfig(Base):
     name: Optional[str] = "self-registration"
     namespace: Optional[str] = None
     values: Optional[Dict[str, Any]] = {}
     account_expiration_days: Optional[int] = 7
     approved_domains: Optional[List[str]] = []
     coupons: Optional[List[str]] = []
-    registration_group: Optional[str] = ""    
+    registration_group: Optional[str] = ""
     affinity: SelfRegistrationConfig = SelfRegistrationConfig()
 
 
 class InputSchema(Base):
     self_registration: SelfRegistrationConfig = SelfRegistrationConfig()
 
+
 class SelfRegistrationStage(NebariTerraformStage):
     name = "self-registration"
     priority = 103
-    wait = True # wait for install to complete on nebari deploy
+    wait = True  # wait for install to complete on nebari deploy
     input_schema = InputSchema
 
     def tf_objects(self) -> List[Dict]:
@@ -52,12 +53,10 @@ def tf_objects(self) -> List[Dict]:
             NebariKubernetesProvider(self.config),
         ]
 
-
-
     @property
     def template_directory(self):
         return Path(inspect.getfile(self.__class__)).parent / "terraform"
-    
+
     def _attempt_keycloak_connection(
         self,
         keycloak_url,
@@ -84,12 +83,10 @@ def _attempt_keycloak_connection(
                     client_id=client_id,
                     verify=verify,
                 )
-                c = realm_admin.get_client_id(CLIENT_NAME) # lookup client guid
-                existing_client = realm_admin.get_client(c) # query client info
+                c = realm_admin.get_client_id(CLIENT_NAME)  # lookup client guid
+                existing_client = realm_admin.get_client(c)  # query client info
                 if existing_client != None and existing_client["name"] == CLIENT_NAME:
-                    print(
-                        f"Attempt {i+1} succeeded connecting to keycloak and nebari client={CLIENT_NAME} exists"
-                    )
+                    print(f"Attempt {i+1} succeeded connecting to keycloak and nebari client={CLIENT_NAME} exists")
                     return True
                 else:
                     print(
@@ -99,28 +96,26 @@ def _attempt_keycloak_connection(
                 print(f"Attempt {i+1} failed connecting to keycloak {client_realm_name} realm -- {e}")
             time.sleep(timeout)
         return False
-    
+
     def check(self, stage_outputs: Dict[str, Dict[str, Any]], disable_prompt=False) -> bool:
-                
+
         try:
             _ = self.config.escaped_project_name
             _ = self.config.provider
-            
+
         except KeyError:
-            print(
-                "\nBase config values not found: escaped_project_name, provider"
-            )
+            print("\nBase config values not found: escaped_project_name, provider")
             return False
 
         keycloak_config = self.get_keycloak_config(stage_outputs)
 
         if not self._attempt_keycloak_connection(
-            keycloak_url = keycloak_config["keycloak_url"],
-            username = keycloak_config["username"],
-            password = keycloak_config["password"],
-            master_realm_name = keycloak_config["master_realm_id"],
-            client_id = keycloak_config["master_client_id"],
-            client_realm_name = keycloak_config["realm_id"],
+            keycloak_url=keycloak_config["keycloak_url"],
+            username=keycloak_config["username"],
+            password=keycloak_config["password"],
+            master_realm_name=keycloak_config["master_realm_id"],
+            client_id=keycloak_config["master_client_id"],
+            client_realm_name=keycloak_config["realm_id"],
             verify=False,
         ):
             print(
@@ -133,10 +128,10 @@ def check(self, stage_outputs: Dict[str, Dict[str, Any]], disable_prompt=False)
 
     def input_vars(self, stage_outputs: Dict[str, Dict[str, Any]]):
         keycloak_config = self.get_keycloak_config(stage_outputs)
-    
+
         try:
             domain = stage_outputs["stages/04-kubernetes-ingress"]["domain"]
-            
+
         except KeyError:
             raise Exception("Prerequisite stage output(s) not found: stages/04-kubernetes-ingress")
 
@@ -163,11 +158,15 @@ def input_vars(self, stage_outputs: Dict[str, Dict[str, Any]]):
             "overrides": self.config.self_registration.values,
             "affinity": {
                 "enabled": self.config.self_registration.affinity.enabled,
-                "selector": self.config.self_registration.affinity.selector.__dict__
-                if isinstance(self.config.self_registration.affinity.selector, SelfRegistrationAffinitySelectorConfig)
-                else self.config.self_registration.affinity.selector,
+                "selector": (
+                    self.config.self_registration.affinity.selector.__dict__
+                    if isinstance(
+                        self.config.self_registration.affinity.selector, SelfRegistrationAffinitySelectorConfig
+                    )
+                    else self.config.self_registration.affinity.selector
+                ),
             },
-
+            "theme": self.config.theme.jupyterhub,
         }
 
     def get_keycloak_config(self, stage_outputs: Dict[str, Dict[str, Any]]):
@@ -182,7 +181,8 @@ def get_keycloak_config(self, stage_outputs: Dict[str, Dict[str, Any]]):
             "master_client_id": stage_outputs[directory]["keycloak_credentials"]["value"]["client_id"],
             "realm_id": stage_outputs["stages/06-kubernetes-keycloak-configuration"]["realm_id"]["value"],
         }
-
+
+
 @hookimpl
 def nebari_stage() -> List[NebariStage]:
-    return [ SelfRegistrationStage ]
+    return [SelfRegistrationStage]