Environ secrets

luxonis_ml/data/parsers/luxonis_parser.py

@@ -285,7 +285,7 @@ def _download_roboflow_dataset(
                 "Please set it to your Roboflow API key."
             )
 
-        rf = Roboflow(api_key=environ.ROBOFLOW_API_KEY)
+        rf = Roboflow(api_key=environ.ROBOFLOW_API_KEY.get_secret_value())
         parts = dataset_dir.split("roboflow://")[1].split("/")
         if len(parts) != 4:
             raise ValueError(

luxonis_ml/utils/environ.py

@@ -2,7 +2,7 @@
 from pathlib import Path
 from typing import Any, Literal, cast
 
-from pydantic import model_serializer
+from pydantic import SecretStr, model_serializer
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 from luxonis_ml.typing import Params
@@ -17,18 +17,18 @@ class Environ(BaseSettings):
         env_file=".env", env_file_encoding="utf-8", extra="ignore"
     )
 
-    AWS_ACCESS_KEY_ID: str | None = None
-    AWS_SECRET_ACCESS_KEY: str | None = None
+    AWS_ACCESS_KEY_ID: SecretStr | None = None
+    AWS_SECRET_ACCESS_KEY: SecretStr | None = None
     AWS_S3_ENDPOINT_URL: str | None = None
 
     MLFLOW_CLOUDFLARE_ID: str | None = None
-    MLFLOW_CLOUDFLARE_SECRET: str | None = None
+    MLFLOW_CLOUDFLARE_SECRET: SecretStr | None = None
     MLFLOW_S3_BUCKET: str | None = None
     MLFLOW_S3_ENDPOINT_URL: str | None = None
     MLFLOW_TRACKING_URI: str | None = None
 
     POSTGRES_USER: str | None = None
-    POSTGRES_PASSWORD: str | None = None
+    POSTGRES_PASSWORD: SecretStr | None = None
     POSTGRES_HOST: str | None = None
     POSTGRES_PORT: str | None = None
     POSTGRES_DB: str | None = None
@@ -38,9 +38,9 @@ class Environ(BaseSettings):
     LUXONISML_TEAM_ID: str = "offline"
     LUXONISML_DISABLE_SETUP_LOGGING: bool = False
 
-    ROBOFLOW_API_KEY: str | None = None
+    ROBOFLOW_API_KEY: SecretStr | None = None
 
-    GOOGLE_APPLICATION_CREDENTIALS: str | None = None
+    GOOGLE_APPLICATION_CREDENTIALS: SecretStr | None = None
 
     LOG_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = (
         "INFO"

luxonis_ml/utils/filesystem.py

@@ -152,8 +152,12 @@ def init_fsspec_filesystem(self) -> fsspec.AbstractFileSystem:
             # NOTE: In theory boto3 should look in environment variables automatically but it doesn't seem to work
             fs = fsspec.filesystem(
                 self.protocol,
-                key=environ.AWS_ACCESS_KEY_ID,
-                secret=environ.AWS_SECRET_ACCESS_KEY,
+                key=environ.AWS_ACCESS_KEY_ID.get_secret_value()
+                if environ.AWS_ACCESS_KEY_ID is not None
+                else None,
+                secret=environ.AWS_SECRET_ACCESS_KEY.get_secret_value()
+                if environ.AWS_SECRET_ACCESS_KEY is not None
+                else None,
                 endpoint_url=environ.AWS_S3_ENDPOINT_URL,
             )
         elif self.protocol == "gcs":

tests/test_utils/test_config.py

@@ -7,7 +7,6 @@
 from pydantic import BaseModel
 
 from luxonis_ml.typing import Params
-from luxonis_ml.utils import environ
 from luxonis_ml.utils.config import LuxonisConfig
 
 CONFIG_DATA = {
@@ -271,19 +270,6 @@ def test_get(config_file: str):
         cfg.get("list_config.index.int_list_param")
 
 
-def test_environ(monkeypatch: pytest.MonkeyPatch):
-    monkeypatch.delenv("POSTGRES_HOST", raising=False)
-    assert environ.POSTGRES_HOST is None
-
-    monkeypatch.setenv("POSTGRES_HOST", "first.example.com")
-    assert environ.POSTGRES_HOST == "first.example.com"
-
-    monkeypatch.setenv("POSTGRES_HOST", "second.example.com")
-    assert environ.POSTGRES_HOST == "second.example.com"
-
-    assert environ.model_dump() == {}
-
-
 def test_safe_load(config_file: str):
     cfg = Config.get_config(config_file)
     with tempfile.NamedTemporaryFile(delete=False) as f:

tests/test_utils/test_environ.py

@@ -0,0 +1,27 @@
+from collections.abc import Generator
+from pathlib import Path
+
+import pytest
+from pydantic import SecretStr
+
+from luxonis_ml.utils.environ import Environ
+
+
+@pytest.fixture
+def dotenv_file(tempdir: Path) -> Generator[Path]:
+    path = tempdir / "dotenv"
+    path.write_text("POSTGRES_USER=test\nPOSTGRES_PASSWORD=pass\n")
+    yield path
+    path.unlink(missing_ok=True)
+
+
+def test_environ(dotenv_file: Path):
+    environ = Environ(
+        POSTGRES_USER="user",
+        _env_file=dotenv_file,  # type: ignore
+    )
+    assert environ.POSTGRES_USER == "user"
+    assert isinstance(environ.POSTGRES_PASSWORD, SecretStr)
+    assert environ.POSTGRES_PASSWORD.get_secret_value() == "pass"
+
+    assert environ.model_dump() == {}