[FL-541] feat(): Added endpoint url customization capability

flow360/cli/app.py

@@ -10,6 +10,7 @@
 
 from flow360.cli import dict_utils
 from flow360.component.project_utils import show_projects_with_keyword_filter
+from flow360.environment import Env
 
 home = expanduser("~")
 # pylint: disable=invalid-name
@@ -41,6 +42,7 @@ def flow360():
 @click.option(
     "--uat", prompt=False, type=bool, is_flag=True, help="Only use this apikey in UAT environment."
 )
+@click.option("--env", prompt=False, default=None, help="Only use this apikey in this environment.")
 @click.option(
     "--suppress-submit-warning",
     type=bool,
@@ -51,8 +53,8 @@ def flow360():
     type=bool,
     help="Toggle beta features support",
 )
-# pylint: disable=too-many-arguments
-def configure(apikey, profile, dev, uat, suppress_submit_warning, beta_features):
+# pylint: disable=too-many-arguments, too-many-branches
+def configure(apikey, profile, dev, uat, env, suppress_submit_warning, beta_features):
     """
     Configure flow360.
     """
@@ -70,6 +72,16 @@ def configure(apikey, profile, dev, uat, suppress_submit_warning, beta_features)
             entry = {profile: {"dev": {"apikey": apikey}}}
         elif uat is True:
             entry = {profile: {"uat": {"apikey": apikey}}}
+        elif env:
+            if env == "dev":
+                raise ValueError("Cannot set dev environment with --env, please use --dev instead.")
+            if env == "uat":
+                raise ValueError("Cannot set uat environment with --env, please use --uat instead.")
+            if env == "prod":
+                raise ValueError(
+                    "Cannot set prod environment with --env, please remove --env and its argument."
+                )
+            entry = {profile: {env: {"apikey": apikey}}}
         else:
             entry = {profile: {"apikey": apikey}}
         dict_utils.merge_overwrite(config, entry)
@@ -98,13 +110,22 @@ def configure(apikey, profile, dev, uat, suppress_submit_warning, beta_features)
 # For displaying all projects
 @click.command("show_projects", context_settings={"show_default": True})
 @click.option("--keyword", "-k", help="Filter projects by keyword", default=None, type=str)
-def show_projects(keyword):
+@click.option("--env", prompt=False, default=None, help="The environment used for the query.")
+def show_projects(keyword, env: str):
     """
     Display all available projects with optional keyword filter.
     """
+    prev_env_config = None
+    if env:
+        env_config = Env.load(env)
+        prev_env_config = Env.current
+        env_config.active()
 
     show_projects_with_keyword_filter(search_keyword=keyword)
 
+    if prev_env_config:
+        prev_env_config.active()
+
 
 flow360.add_command(configure)
 flow360.add_command(show_projects)

flow360/cloud/s3_utils.py

@@ -139,13 +139,17 @@ def get_client(self):
         :return:
         """
         # pylint: disable=no-member
-        return boto3.client(
-            "s3",
-            region_name=Env.current.aws_region,
-            aws_access_key_id=self.user_credential.access_key_id,
-            aws_secret_access_key=self.user_credential.secret_access_key,
-            aws_session_token=self.user_credential.session_token,
-        )
+        kwargs = {
+            "region_name": Env.current.aws_region,
+            "aws_access_key_id": self.user_credential.access_key_id,
+            "aws_secret_access_key": self.user_credential.secret_access_key,
+            "aws_session_token": self.user_credential.session_token,
+        }
+
+        if Env.current.s3_endpoint_url is not None:
+            kwargs["endpoint_url"] = Env.current.s3_endpoint_url
+
+        return boto3.client("s3", **kwargs)
 
     def is_expired(self):
         """
@@ -407,4 +411,4 @@ def _get_s3_sts_token(self, resource_id: str, file_name: str) -> _S3STSToken:
         )
 
 
-_s3_sts_tokens: [str, _S3STSToken] = {}
+_s3_sts_tokens: dict[str, _S3STSToken] = {}

flow360/environment.py

@@ -4,8 +4,15 @@
 
 from __future__ import annotations
 
+import os
+from typing import Optional
+
+import toml
 from pydantic import BaseModel
 
+from .file_path import flow360_dir
+from .log import log
+
 
 class EnvironmentConfig(BaseModel):
     """
@@ -18,7 +25,8 @@ class EnvironmentConfig(BaseModel):
     web_url: str
     aws_region: str
     apikey_profile: str
-    portal_web_api_endpoint: str = None
+    portal_web_api_endpoint: Optional[str] = None
+    s3_endpoint_url: Optional[str] = None
 
     @classmethod
     def from_domain(cls, name, domain, aws_region, apikey_profile="default") -> EnvironmentConfig:
@@ -82,6 +90,54 @@ def get_web_real_url(self, path: str):
         """
         return "/".join([self.web_url, path])
 
+    @classmethod
+    def from_config(cls, env_config_name: str):
+        """
+        Load specified environment configuration from config.toml.
+        """
+        config_file = os.path.join(flow360_dir, "config.toml")
+        if os.path.exists(config_file):
+            with open(config_file, encoding="utf-8", mode="r") as file_handler:
+                config = toml.loads(file_handler.read())
+                if "env_config" not in config:
+                    raise ValueError("Cannot find any environment configuration in config file.")
+
+                if env_config_name in config["env_config"]:
+                    env_config = config["env_config"][env_config_name]
+                    log.info(
+                        f"Loaded environment configuration from config.toml for {env_config_name}."
+                    )
+                    return cls.model_validate(env_config)
+
+                raise ValueError(
+                    f"Environment configuration for `{env_config_name}` not found."
+                    f" Available: {config['env_config'].keys()}."
+                )
+        else:
+            raise FileNotFoundError("Failed to find the config file.")
+
+    def save_config(self):
+        """
+        Save the configuration to the config file.
+        """
+        current_env_config = self.model_dump(mode="json")
+        config_file = os.path.join(flow360_dir, "config.toml")
+        if os.path.exists(config_file):
+            with open(config_file, encoding="utf-8", mode="r") as file_handler:
+                existing_config = toml.loads(file_handler.read())
+                if "env_config" not in existing_config:
+                    existing_config["env_config"] = {}
+                existing_config["env_config"][self.name] = current_env_config
+            with open(config_file, encoding="utf-8", mode="w") as file_handler:
+                file_handler.write(toml.dumps(existing_config))
+        else:
+            # Create the config file if it doesn't exist.
+            # This will not be triggered most likely.
+            log.info("Creating config.toml since it does not exist.")
+            with open(config_file, encoding="utf-8", mode="w") as file_handler:
+                file_handler.write(toml.dumps({"env_config": {self.name: [current_env_config]}}))
+        log.info("Saved environment configuration to config.toml.")
+
 
 dev = EnvironmentConfig.from_domain(
     name="dev", domain="dev-simulation.cloud", aws_region="us-east-1", apikey_profile="dev"
@@ -161,6 +217,22 @@ def preprod(self):
         """
         return preprod
 
+    def load(self, /, env_config_name: str):
+        """
+        Load the environment configuration from config.toml.
+        """
+        if not isinstance(env_config_name, str):
+            raise ValueError(
+                f"The name of environment setting must be a string. Instead got {env_config_name}."
+            )
+
+        predefined_envs = (dev, uat, prod, preprod)
+        for env in predefined_envs:
+            if env_config_name == env.name:
+                return env
+
+        return EnvironmentConfig.from_config(env_config_name)
+
     def set_current(self, config: EnvironmentConfig):
         """
         Set the current environment.

flow360/user_config.py

@@ -6,6 +6,7 @@
 
 import toml
 
+from .environment import prod
 from .file_path import flow360_dir
 from .log import log
 
@@ -77,10 +78,13 @@ def apikey(self, env):
             return self._apikey
         # Check if environment-specific apikey exists
         key = self.config.get(self.profile, {})
-        if key and env.name == "dev":
-            key = key.get("dev")
-        elif key and env.name == "uat":
-            key = key.get("uat")
+
+        if env.name != prod.name:
+            # By default the production environment is used.
+            # If other environment is used, check if the key exists
+            key = key.get(env.name, None)
+        if key is None:
+            log.warning(f"Cannot find api key associated with environment '{env.name}'.")
         return None if key is None else key.get("apikey", "")
 
     def suppress_submit_warning(self):