refresh token function

core/digitalhub_core/__init__.py

@@ -50,4 +50,4 @@
     update_workflow,
 )
 from digitalhub_core.stores.builder import set_store
-from digitalhub_core.utils.env_utils import set_dhcore_env
+from digitalhub_core.utils.env_utils import refresh_token, set_dhcore_env

core/digitalhub_core/client/objects/dhcore.py

@@ -478,6 +478,9 @@ def _configure(self, config: dict | None = None) -> None:
         -------
         None
         """
+        # Load env from file
+        self._load_env()
+
         self._get_endpoints_from_env()
 
         if config is not None:
@@ -499,6 +502,9 @@ def _configure(self, config: dict | None = None) -> None:
 
         self._get_auth_from_env()
 
+        # Propagate access and refresh token to env file
+        self._write_env()
+
     def _get_endpoints_from_env(self) -> None:
         """
         Get the DHCore endpoint and token issuer endpoint from env.
@@ -512,8 +518,6 @@ def _get_endpoints_from_env(self) -> None:
         Exception
             If the endpoint of DHCore is not set in the env variables.
         """
-        self._load_env()
-
         core_endpt = os.getenv("DHCORE_ENDPOINT")
         if core_endpt is None:
             raise BackendError("Endpoint not set as environment variables.")
@@ -602,7 +606,7 @@ def _get_refresh_endpoint(self) -> str:
 
         # Call
         r = request("GET", url, timeout=60)
-        r.raise_for_status()
+        self._raise_for_error(r)
         return r.json().get("token_endpoint")
 
     def _call_refresh_token_endpoint(self, url: str) -> dict:
@@ -627,7 +631,7 @@ def _call_refresh_token_endpoint(self, url: str) -> dict:
         }
         headers = {"Content-Type": "application/x-www-form-urlencoded"}
         r = request("POST", url, data=payload, headers=headers, timeout=60)
-        r.raise_for_status()
+        self._raise_for_error(r)
         return r.json()
 
     @staticmethod

core/digitalhub_core/entities/project/entity/_base.py

@@ -164,13 +164,10 @@ def _export_not_embedded(self, obj: dict) -> dict:
         """
         # Cycle over entity types
         for entity_type in CTX_ENTITIES:
-
             # Entity types are stored as a list of entities
             for idx, entity in enumerate(obj.get("spec", {}).get(entity_type, [])):
-
                 # Export entity if not embedded is in metadata, else do nothing
                 if not entity["metadata"]["embedded"]:
-
                     # Get entity object from backend
                     obj_dict: dict = read_entity_api_ctx(entity["key"])
 

core/digitalhub_core/utils/env_utils.py

@@ -99,6 +99,18 @@ def update_client_from_env() -> None:
         client._auth_type = "basic"
 
 
+def refresh_token() -> None:
+    """
+    Function to refresh token.
+
+    Returns
+    -------
+    None
+    """
+    client: ClientDHCore = get_client(local=False)
+    client._get_new_access_token()
+
+
 def get_s3_bucket() -> str | None:
     """
     Function to get S3 bucket name.

digitalhub/__init__.py

@@ -35,6 +35,7 @@
     new_secret,
     new_task,
     new_workflow,
+    refresh_token,
     set_dhcore_env,
     set_store,
     update_artifact,

runtimes/kfp/digitalhub_runtime_kfp/utils/functions.py

@@ -154,7 +154,7 @@ def _kfp_build_execution(pipeline: Callable, function_args) -> dict:
         # need to replicate the build
         workflow = build_kfp_pipeline(run, pipeline)
 
-        run_status = { "state": State.COMPLETED.value, "results": {"workflow":  workflow} }
+        run_status = {"state": State.COMPLETED.value, "results": {"workflow": workflow}}
         _update_status(run.get("key"), run_status)
 
         return run_status