Allows Secret groups to be optional and configurable

flytekit/configuration/plugin.py

@@ -39,6 +39,10 @@ def get_remote(
     def configure_pyflyte_cli(main: Command) -> Command:
         """Configure pyflyte's CLI."""
 
+    @staticmethod
+    def secret_requires_group() -> bool:
+        """Return True if secrets require group entry."""
+
 
 class FlytekitPlugin:
     @staticmethod
@@ -62,6 +66,11 @@ def configure_pyflyte_cli(main: Command) -> Command:
         """Configure pyflyte's CLI."""
         return main
 
+    @staticmethod
+    def secret_requires_group() -> bool:
+        """Return True if secrets require group entry."""
+        return True
+
 
 def _get_plugin_from_entrypoint():
     """Get plugin from entrypoint."""

flytekit/core/context_manager.py

@@ -350,7 +350,11 @@ def __getattr__(self, item: str) -> _GroupSecrets:
         return self._GroupSecrets(item, self)
 
     def get(
-        self, group: str, key: Optional[str] = None, group_version: Optional[str] = None, encode_mode: str = "r"
+        self,
+        group: Optional[str] = None,
+        key: Optional[str] = None,
+        group_version: Optional[str] = None,
+        encode_mode: str = "r",
     ) -> str:
         """
         Retrieves a secret using the resolution order -> Env followed by file. If not found raises a ValueError
@@ -370,15 +374,19 @@ def get(
             f"in Env Var:{env_var} and FilePath: {fpath}"
         )
 
-    def get_secrets_env_var(self, group: str, key: Optional[str] = None, group_version: Optional[str] = None) -> str:
+    def get_secrets_env_var(
+        self, group: Optional[str] = None, key: Optional[str] = None, group_version: Optional[str] = None
+    ) -> str:
         """
         Returns a string that matches the ENV Variable to look for the secrets
         """
         self.check_group_key(group)
         l = [k.upper() for k in filter(None, (group, group_version, key))]
         return f"{self._env_prefix}{'_'.join(l)}"
 
-    def get_secrets_file(self, group: str, key: Optional[str] = None, group_version: Optional[str] = None) -> str:
+    def get_secrets_file(
+        self, group: Optional[str] = None, key: Optional[str] = None, group_version: Optional[str] = None
+    ) -> str:
         """
         Returns a path that matches the file to look for the secrets
         """
@@ -388,8 +396,10 @@ def get_secrets_file(self, group: str, key: Optional[str] = None, group_version:
         return os.path.join(self._base_dir, *l)
 
     @staticmethod
-    def check_group_key(group: str):
-        if group is None or group == "":
+    def check_group_key(group: Optional[str]):
+        from flytekit.configuration.plugin import get_plugin
+
+        if get_plugin().secret_requires_group() and (group is None or group == ""):
             raise ValueError("secrets group is a mandatory field.")
 
 

flytekit/models/security.py

@@ -35,13 +35,15 @@ class MountType(Enum):
         Caution: May not be supported in all environments
         """
 
-    group: str
+    group: Optional[str] = None
     key: Optional[str] = None
     group_version: Optional[str] = None
     mount_requirement: MountType = MountType.ANY
 
     def __post_init__(self):
-        if self.group is None:
+        from flytekit.configuration.plugin import get_plugin
+
+        if get_plugin().secret_requires_group() and self.group is None:
             raise ValueError("Group is a required parameter")
 
     def to_flyte_idl(self) -> _sec.Secret:

tests/flytekit/unit/core/test_context_manager.py

@@ -1,11 +1,14 @@
 import base64
 import os
 from datetime import datetime
+from pathlib import Path
+from unittest.mock import Mock
 
 import mock
 import py
 import pytest
 
+import flytekit.configuration.plugin
 from flytekit.configuration import (
     SERIALIZED_CONTEXT_ENV_VAR,
     FastSerializationSettings,
@@ -128,6 +131,24 @@ def test_secrets_manager_get_envvar():
     assert sec.get_secrets_env_var("group") == f"{cfg.env_prefix}GROUP"
 
 
+def test_secret_manager_no_group(monkeypatch):
+    plugin_mock = Mock()
+    plugin_mock.secret_requires_group.return_value = False
+    mock_global_plugin = {"plugin": plugin_mock}
+    monkeypatch.setattr(flytekit.configuration.plugin, "_GLOBAL_CONFIG", mock_global_plugin)
+
+    sec = SecretsManager()
+    cfg = SecretsConfig.auto()
+    sec.check_group_key(None)
+    sec.check_group_key("")
+
+    assert sec.get_secrets_env_var(key="ABC") == f"{cfg.env_prefix}ABC"
+
+    default_path = Path(cfg.default_dir)
+    expected_path = default_path / f"{cfg.file_prefix}abc"
+    assert sec.get_secrets_file(key="ABC") == str(expected_path)
+
+
 def test_secrets_manager_get_file():
     sec = SecretsManager()
     with pytest.raises(ValueError):

tests/flytekit/unit/models/core/test_security.py

@@ -1,3 +1,6 @@
+from unittest.mock import Mock
+
+import flytekit.configuration.plugin
 from flytekit.models.security import Secret
 
 
@@ -11,3 +14,13 @@ def test_secret():
     obj2 = Secret.from_flyte_idl(obj.to_flyte_idl())
     assert obj2.key is None
     assert obj2.group_version == "v1"
+
+
+def test_secret_no_group(monkeypatch):
+    plugin_mock = Mock()
+    plugin_mock.secret_requires_group.return_value = False
+    mock_global_plugin = {"plugin": plugin_mock}
+    monkeypatch.setattr(flytekit.configuration.plugin, "_GLOBAL_CONFIG", mock_global_plugin)
+
+    s = Secret(key="key")
+    assert s.group is None