environ secrets

Author: kozlov721

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"

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,37 @@
+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(
+        "AWS_ACCESS_KEY_ID=example_access_key\n"
+        "AWS_SECRET_ACCESS_KEY=example_secret_key\n"
+        "AWS_S3_ENDPOINT_URL=http://example.com\n"
+    )
+    yield path
+    path.unlink(missing_ok=True)
+
+
+def test_environ(dotenv_file: Path, monkeypatch: pytest.MonkeyPatch):
+    environ = Environ(
+        POSTGRES_USER="user",
+        _env_file=dotenv_file,  # type: ignore
+    )
+    assert environ.POSTGRES_USER == "user"
+    assert isinstance(environ.AWS_ACCESS_KEY_ID, SecretStr)
+    assert isinstance(environ.AWS_SECRET_ACCESS_KEY, SecretStr)
+    assert (
+        environ.AWS_SECRET_ACCESS_KEY.get_secret_value()
+        == "example_secret_key"
+    )
+    assert environ.AWS_S3_ENDPOINT_URL == "http://example.com"
+    assert environ.AWS_ACCESS_KEY_ID.get_secret_value() == "example_access_key"
+
+    assert environ.model_dump() == {}