move repeated logic

Author: jonhealy1

stac_fastapi/core/stac_fastapi/core/utilities.py

@@ -12,20 +12,75 @@
 MAX_LIMIT = 10000
 
 
-def get_bool_env(name: str, default: bool = False) -> bool:
+def validate_refresh(value: Union[str, bool]) -> str:
+    """
+    Validate the `refresh` parameter value.
+
+    Args:
+        value (Union[str, bool]): The `refresh` parameter value, which can be a string or a boolean.
+
+    Returns:
+        str: The validated value of the `refresh` parameter, which can be "true", "false", or "wait_for".
+    """
+    logger = logging.getLogger(__name__)
+
+    # Handle boolean-like values using get_bool_env
+    if isinstance(value, bool) or value in {
+        "true",
+        "false",
+        "1",
+        "0",
+        "yes",
+        "no",
+        "y",
+        "n",
+    }:
+        is_true = get_bool_env("DATABASE_REFRESH", default=value)
+        return "true" if is_true else "false"
+
+    # Normalize to lowercase for case-insensitivity
+    value = value.lower()
+
+    # Handle "wait_for" explicitly
+    if value == "wait_for":
+        return "wait_for"
+
+    # Log a warning for invalid values and default to "false"
+    logger.warning(
+        f"Invalid value for `refresh`: '{value}'. Expected 'true', 'false', or 'wait_for'. Defaulting to 'false'."
+    )
+    return "false"
+
+
+def get_bool_env(name: str, default: Union[bool, str] = False) -> bool:
     """
     Retrieve a boolean value from an environment variable.
 
     Args:
         name (str): The name of the environment variable.
-        default (bool, optional): The default value to use if the variable is not set or unrecognized. Defaults to False.
+        default (Union[bool, str], optional): The default value to use if the variable is not set or unrecognized. Defaults to False.
 
     Returns:
         bool: The boolean value parsed from the environment variable.
     """
-    value = os.getenv(name, str(default).lower())
     true_values = ("true", "1", "yes", "y")
     false_values = ("false", "0", "no", "n")
+
+    # Normalize the default value
+    if isinstance(default, bool):
+        default_str = "true" if default else "false"
+    elif isinstance(default, str):
+        default_str = default.lower()
+    else:
+        logger = logging.getLogger(__name__)
+        logger.warning(
+            f"The `default` parameter must be a boolean or string, got {type(default).__name__}. "
+            f"Falling back to `False`."
+        )
+        default_str = "false"
+
+    # Retrieve and normalize the environment variable value
+    value = os.getenv(name, default_str)
     if value.lower() in true_values:
         return True
     elif value.lower() in false_values:
@@ -34,9 +89,9 @@ def get_bool_env(name: str, default: bool = False) -> bool:
         logger = logging.getLogger(__name__)
         logger.warning(
             f"Environment variable '{name}' has unrecognized value '{value}'. "
-            f"Expected one of {true_values + false_values}. Using default: {default}"
+            f"Expected one of {true_values + false_values}. Using default: {default_str}"
         )
-        return default
+        return default_str in true_values
 
 
 def resolve_refresh(refresh: str) -> str:

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/config.py

@@ -10,7 +10,7 @@
 
 from elasticsearch import Elasticsearch  # type: ignore[attr-defined]
 from stac_fastapi.core.base_settings import ApiBaseSettings
-from stac_fastapi.core.utilities import get_bool_env
+from stac_fastapi.core.utilities import get_bool_env, validate_refresh
 from stac_fastapi.types.config import ApiSettings
 
 
@@ -96,15 +96,8 @@ def database_refresh(self) -> Union[bool, str]:
         Returns:
             Union[bool, str]: The value of DATABASE_REFRESH, which can be True, False, or "wait_for".
         """
-        value = os.getenv("DATABASE_REFRESH", "false").lower()
-        if value in {"true", "false"}:
-            return value == "true"
-        elif value == "wait_for":
-            return "wait_for"
-        else:
-            raise ValueError(
-                "Invalid value for DATABASE_REFRESH. Must be 'true', 'false', or 'wait_for'."
-            )
+        value = os.getenv("DATABASE_REFRESH", "false")
+        return validate_refresh(value)
 
     @property
     def create_client(self):
@@ -135,15 +128,8 @@ def database_refresh(self) -> Union[bool, str]:
         Returns:
             Union[bool, str]: The value of DATABASE_REFRESH, which can be True, False, or "wait_for".
         """
-        value = os.getenv("DATABASE_REFRESH", "false").lower()
-        if value in {"true", "false"}:
-            return value == "true"
-        elif value == "wait_for":
-            return "wait_for"
-        else:
-            raise ValueError(
-                "Invalid value for DATABASE_REFRESH. Must be 'true', 'false', or 'wait_for'."
-            )
+        value = os.getenv("DATABASE_REFRESH", "false")
+        return validate_refresh(value)
 
     @property
     def create_client(self):

stac_fastapi/opensearch/stac_fastapi/opensearch/config.py

@@ -8,7 +8,7 @@
 from opensearchpy import AsyncOpenSearch, OpenSearch
 
 from stac_fastapi.core.base_settings import ApiBaseSettings
-from stac_fastapi.core.utilities import get_bool_env
+from stac_fastapi.core.utilities import get_bool_env, validate_refresh
 from stac_fastapi.types.config import ApiSettings
 
 
@@ -93,15 +93,8 @@ def database_refresh(self) -> Union[bool, str]:
         Returns:
             Union[bool, str]: The value of DATABASE_REFRESH, which can be True, False, or "wait_for".
         """
-        value = os.getenv("DATABASE_REFRESH", "false").lower()
-        if value in {"true", "false"}:
-            return value == "true"
-        elif value == "wait_for":
-            return "wait_for"
-        else:
-            raise ValueError(
-                "Invalid value for DATABASE_REFRESH. Must be 'true', 'false', or 'wait_for'."
-            )
+        value = os.getenv("DATABASE_REFRESH", "false")
+        return validate_refresh(value)
 
     @property
     def create_client(self):
@@ -132,15 +125,8 @@ def database_refresh(self) -> Union[bool, str]:
         Returns:
             Union[bool, str]: The value of DATABASE_REFRESH, which can be True, False, or "wait_for".
         """
-        value = os.getenv("DATABASE_REFRESH", "false").lower()
-        if value in {"true", "false"}:
-            return value == "true"
-        elif value == "wait_for":
-            return "wait_for"
-        else:
-            raise ValueError(
-                "Invalid value for DATABASE_REFRESH. Must be 'true', 'false', or 'wait_for'."
-            )
+        value = os.getenv("DATABASE_REFRESH", "false")
+        return validate_refresh(value)
 
     @property
     def create_client(self):

stac_fastapi/tests/config/test_config_settings.py

@@ -44,15 +44,15 @@ def test_database_refresh_true(monkeypatch):
     monkeypatch.setenv("DATABASE_REFRESH", "true")
     settings_class, _ = get_settings_class()
     settings = settings_class()
-    assert settings.database_refresh is True
+    assert settings.database_refresh == "true"
 
 
 def test_database_refresh_false(monkeypatch):
     """Test that DATABASE_REFRESH env var disables database refresh."""
     monkeypatch.setenv("DATABASE_REFRESH", "false")
     settings_class, _ = get_settings_class()
     settings = settings_class()
-    assert settings.database_refresh is False
+    assert settings.database_refresh == "false"
 
 
 def test_database_refresh_wait_for(monkeypatch):