use get bool env method

Author: jonhealy1

CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Added
 - Added support for high-performance direct response mode for both Elasticsearch and Opensearch backends, controlled by the `ENABLE_DIRECT_RESPONSE` environment variable. When enabled (`ENABLE_DIRECT_RESPONSE=true`), endpoints return Starlette Response objects directly, bypassing FastAPI's jsonable_encoder and Pydantic serialization for significantly improved performance on large search responses. **Note:** In this mode, all FastAPI dependencies (including authentication, custom status codes, and validation) are disabled for all routes. Default is `false` for safety. A warning is logged at startup if enabled. See [issue #347](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/issues/347) and [PR #359](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/359).
+- Added robust tests for the `ENABLE_DIRECT_RESPONSE` environment variable, covering both Elasticsearch and OpenSearch backends. Tests gracefully handle missing backends by attempting to import both configs and skipping if neither is available. [#359](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/359)
 
 ### Changed
 - Updated test suite to use `httpx.ASGITransport(app=...)` for FastAPI app testing (removes deprecation warning). [#359](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/359)
@@ -25,6 +26,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - Updated all Pydantic models to use `ConfigDict` instead of class-based `Config` for Pydantic v2 compatibility. This resolves deprecation warnings and prepares for Pydantic v3. [#359](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/359)
 - Migrated all Pydantic `@root_validator` validators to `@model_validator` for Pydantic v2 compatibility. [#359](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/359)
 - Migrated startup event handling from deprecated `@app.on_event("startup")` to FastAPI's recommended lifespan context manager. This removes deprecation warnings and ensures compatibility with future FastAPI versions. [#359](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/359)
+- Refactored all boolean environment variable parsing in both Elasticsearch and OpenSearch backends to use the shared `get_bool_env` utility. This ensures robust and consistent handling of environment variables such as `ES_USE_SSL`, `ES_HTTP_COMPRESS`, and `ES_VERIFY_CERTS` across both backends. [#359](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/359)
+
 
 ### Fixed
 

stac_fastapi/core/stac_fastapi/core/utilities.py

@@ -3,13 +3,42 @@
 This module contains functions for transforming geospatial coordinates,
 such as converting bounding boxes to polygon representations.
 """
+import logging
+import os
 from typing import Any, Dict, List, Optional, Set, Union
 
 from stac_fastapi.types.stac import Item
 
 MAX_LIMIT = 10000
 
 
+def get_bool_env(name: str, default: bool = 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.
+
+    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")
+    if value.lower() in true_values:
+        return True
+    elif value.lower() in false_values:
+        return False
+    else:
+        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}"
+        )
+        return default
+
+
 def bbox2polygon(b0: float, b1: float, b2: float, b3: float) -> List[List[List[float]]]:
     """Transform a bounding box represented by its four coordinates `b0`, `b1`, `b2`, and `b3` into a polygon.
 

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/config.py

@@ -9,12 +9,13 @@
 
 from elasticsearch import AsyncElasticsearch, Elasticsearch  # type: ignore
 from stac_fastapi.core.base_settings import ApiBaseSettings
+from stac_fastapi.core.utilities import get_bool_env
 from stac_fastapi.types.config import ApiSettings
 
 
 def _es_config() -> Dict[str, Any]:
     # Determine the scheme (http or https)
-    use_ssl = os.getenv("ES_USE_SSL", "true").lower() == "true"
+    use_ssl = get_bool_env("ES_USE_SSL", default=True)
     scheme = "https" if use_ssl else "http"
 
     # Configure the hosts parameter with the correct scheme
@@ -45,7 +46,7 @@ def _es_config() -> Dict[str, Any]:
 
         config["headers"] = headers
 
-    http_compress = os.getenv("ES_HTTP_COMPRESS", "true").lower() == "true"
+    http_compress = get_bool_env("ES_HTTP_COMPRESS", default=True)
     if http_compress:
         config["http_compress"] = True
 
@@ -55,7 +56,7 @@ def _es_config() -> Dict[str, Any]:
 
     # Include SSL settings if using https
     config["ssl_version"] = ssl.TLSVersion.TLSv1_3  # type: ignore
-    config["verify_certs"] = os.getenv("ES_VERIFY_CERTS", "true").lower() != "false"  # type: ignore
+    config["verify_certs"] = get_bool_env("ES_VERIFY_CERTS", default=True)  # type: ignore
 
     # Include CA Certificates if verifying certs
     if config["verify_certs"]:
@@ -83,9 +84,7 @@ class ElasticsearchSettings(ApiSettings, ApiBaseSettings):
     forbidden_fields: Set[str] = _forbidden_fields
     indexed_fields: Set[str] = {"datetime"}
     enable_response_models: bool = False
-    enable_direct_response: bool = (
-        os.getenv("ENABLE_DIRECT_RESPONSE", "false").lower() == "true"
-    )
+    enable_direct_response: bool = get_bool_env("ENABLE_DIRECT_RESPONSE", default=False)
 
     @property
     def create_client(self):
@@ -113,9 +112,7 @@ class AsyncElasticsearchSettings(ApiSettings, ApiBaseSettings):
     forbidden_fields: Set[str] = _forbidden_fields
     indexed_fields: Set[str] = {"datetime"}
     enable_response_models: bool = False
-    enable_direct_response: bool = (
-        os.getenv("ENABLE_DIRECT_RESPONSE", "false").lower() == "true"
-    )
+    enable_direct_response: bool = get_bool_env("ENABLE_DIRECT_RESPONSE", default=False)
 
     @property
     def create_client(self):

stac_fastapi/opensearch/stac_fastapi/opensearch/config.py

@@ -8,12 +8,13 @@
 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.types.config import ApiSettings
 
 
 def _es_config() -> Dict[str, Any]:
     # Determine the scheme (http or https)
-    use_ssl = os.getenv("ES_USE_SSL", "true").lower() == "true"
+    use_ssl = get_bool_env("ES_USE_SSL", default=True)
     scheme = "https" if use_ssl else "http"
 
     # Configure the hosts parameter with the correct scheme
@@ -34,7 +35,7 @@ def _es_config() -> Dict[str, Any]:
         "headers": {"accept": "application/json", "Content-Type": "application/json"},
     }
 
-    http_compress = os.getenv("ES_HTTP_COMPRESS", "true").lower() == "true"
+    http_compress = get_bool_env("ES_HTTP_COMPRESS", default=True)
     if http_compress:
         config["http_compress"] = True
 
@@ -44,7 +45,7 @@ def _es_config() -> Dict[str, Any]:
 
     # Include SSL settings if using https
     config["ssl_version"] = ssl.PROTOCOL_SSLv23  # type: ignore
-    config["verify_certs"] = os.getenv("ES_VERIFY_CERTS", "true").lower() != "false"  # type: ignore
+    config["verify_certs"] = get_bool_env("ES_VERIFY_CERTS", default=True)  # type: ignore
 
     # Include CA Certificates if verifying certs
     if config["verify_certs"]:
@@ -81,9 +82,7 @@ class OpensearchSettings(ApiSettings, ApiBaseSettings):
     forbidden_fields: Set[str] = _forbidden_fields
     indexed_fields: Set[str] = {"datetime"}
     enable_response_models: bool = False
-    enable_direct_response: bool = (
-        os.getenv("ENABLE_DIRECT_RESPONSE", "false").lower() == "true"
-    )
+    enable_direct_response: bool = get_bool_env("ENABLE_DIRECT_RESPONSE", default=False)
 
     @property
     def create_client(self):

stac_fastapi/tests/elasticsearch/test_direct_response.py

@@ -0,0 +1,39 @@
+import importlib
+
+import pytest
+
+
+def get_settings_class():
+    """
+    Try to import ElasticsearchSettings or OpenSearchSettings, whichever is available.
+    Returns a tuple: (settings_class, config_module)
+    """
+    try:
+        config = importlib.import_module("stac_fastapi.elasticsearch.config")
+        importlib.reload(config)
+        return config.ElasticsearchSettings, config
+    except ModuleNotFoundError:
+        try:
+            config = importlib.import_module("stac_fastapi.opensearch.config")
+            importlib.reload(config)
+            return config.OpensearchSettings, config
+        except ModuleNotFoundError:
+            pytest.skip(
+                "Neither Elasticsearch nor OpenSearch config module is available."
+            )
+
+
+def test_enable_direct_response_true(monkeypatch):
+    """Test that ENABLE_DIRECT_RESPONSE env var enables direct response config."""
+    monkeypatch.setenv("ENABLE_DIRECT_RESPONSE", "true")
+    settings_class, _ = get_settings_class()
+    settings = settings_class()
+    assert settings.enable_direct_response is True
+
+
+def test_enable_direct_response_false(monkeypatch):
+    """Test that ENABLE_DIRECT_RESPONSE env var disables direct response config."""
+    monkeypatch.setenv("ENABLE_DIRECT_RESPONSE", "false")
+    settings_class, _ = get_settings_class()
+    settings = settings_class()
+    assert settings.enable_direct_response is False