Merge pull request #2302 from PrefectHQ/minor-secrets-refactor

Minor secrets refactor

CHANGELOG.md

@@ -10,11 +10,11 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 
 ### Enhancements
 
-- None
+- Allow for dynamically changing secret names at runtime - [#2302](https://github.com/PrefectHQ/prefect/pull/2302)
 
 ### Task Library
 
-- None
+- Rename the base secret tasks for clarity - [#2302](https://github.com/PrefectHQ/prefect/pull/2302)
 
 ### Fixes
 
@@ -29,7 +29,7 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 
 ### Breaking Changes
 
-- None
+- Remove `env_var` initialization from `EnvVarSecret` in favor of `name` - [#2302](https://github.com/PrefectHQ/prefect/pull/2302)
 
 ### Contributors
 

docs/outline.toml

@@ -312,7 +312,7 @@ classes = ["ParseRSSFeed"]
 [pages.tasks.secrets]
 title = "Secret Tasks"
 module = "prefect.tasks.secrets"
-classes = ["Secret", "EnvVarSecret"]
+classes = ["SecretBase", "PrefectSecret", "EnvVarSecret"]
 
 [pages.tasks.snowflake]
 title = "Snowflake Tasks"

src/prefect/tasks/secrets/__init__.py

@@ -5,5 +5,5 @@
 class for interacting with other secret providers. Secrets always use a special kind of result handler that
 prevents the persistence of sensitive information.
 """
-from .base import Secret
+from .base import SecretBase, PrefectSecret, Secret
 from .env_var import EnvVarSecret

src/prefect/tasks/secrets/base.py

@@ -1,18 +1,40 @@
 import datetime
+import warnings
 
 from prefect.client.secrets import Secret as _Secret
 from prefect.core.task import Task
 from prefect.engine.result_handlers import SecretResultHandler
+from prefect.utilities.tasks import defaults_from_attrs
 
 
-class Secret(Task):
+class SecretBase(Task):
     """
-    Base Prefect Secrets Task.  This task retrieves the underlying secret through
-    the Prefect Secrets API (which has the ability to toggle between local vs. Cloud secrets).
+    Base Secrets Task.  This task does not perform any action but rather serves as the base
+    task class which should be inherited from when writing new Secret Tasks.
+
     Users should subclass this Task and override its `run` method for plugging into other Secret stores,
     as it is handled differently during execution to ensure the underlying secret value is not accidentally
     persisted in a non-safe location.
 
+    Args:
+        - **kwargs (Any, optional): additional keyword arguments to pass to the Task constructor
+
+    Raises:
+        - ValueError: if a `result_handler` keyword is passed
+    """
+
+    def __init__(self, **kwargs):
+        if kwargs.get("result_handler"):
+            raise ValueError("Result Handlers for Secrets are not configurable.")
+        kwargs["result_handler"] = SecretResultHandler(secret_task=self)
+        super().__init__(**kwargs)
+
+
+class PrefectSecret(Task):
+    """
+    Prefect Secrets Task.  This task retrieves the underlying secret through
+    the Prefect Secrets API (which has the ability to toggle between local vs. Cloud secrets).
+
     Args:
         - name (str): The name of the underlying secret
         - **kwargs (Any, optional): additional keyword arguments to pass to the Task constructor
@@ -30,14 +52,28 @@ def __init__(self, name, **kwargs):
         kwargs["result_handler"] = SecretResultHandler(secret_task=self)
         super().__init__(**kwargs)
 
-    def run(self):
+    @defaults_from_attrs("name")
+    def run(self, name: str = None):
         """
         The run method for Secret Tasks.  This method actually retrieves and returns the underlying secret value
         using the `Secret.get()` method.  Note that this method first checks context for the secret value, and if not
         found either raises an error or queries Prefect Cloud, depending on whether `config.cloud.use_local_secrets`
         is `True` or `False`.
 
+        Args:
+            - name (str, optional): the name of the underlying Secret to retrieve. Defaults
+                to the name provided at initialization.
+
         Returns:
             - Any: the underlying value of the Prefect Secret
         """
-        return _Secret(self.name).get()
+        return _Secret(name).get()
+
+
+class Secret(PrefectSecret):
+    def __init__(self, *args, **kwargs):
+        warnings.warn(
+            "The `Secret` task is deprecated and has been renamed `PrefectSecret`.",
+            UserWarning,
+        )
+        super().__init__(*args, **kwargs)

src/prefect/tasks/secrets/env_var.py

@@ -2,15 +2,15 @@
 from typing import Any, Callable
 
 from prefect.tasks.secrets import Secret
+from prefect.utilities.tasks import defaults_from_attrs
 
 
 class EnvVarSecret(Secret):
     """
     A `Secret` task that retrieves a value from an environment variable.
 
     Args:
-        - env_var (str): the environment variable that contains the secret value
-        - name (str, optional): a name for the task. If not provided, `env_var` is used.
+        - name (str): the environment variable that contains the secret value
         - cast (Callable[[Any], Any]): A function that will be called on the Parameter
             value to coerce it to a type.
         - raise_if_missing (bool): if True, an error will be raised if the env var is not found.
@@ -19,30 +19,33 @@ class EnvVarSecret(Secret):
 
     def __init__(
         self,
-        env_var: str,
-        name: str = None,
+        name: str,
         cast: Callable[[Any], Any] = None,
         raise_if_missing: bool = False,
         **kwargs
     ):
-        self.env_var = env_var
         self.cast = cast
         self.raise_if_missing = raise_if_missing
-        if name is None:
-            name = env_var
-
         super().__init__(name=name, **kwargs)
 
-    def run(self):
+    @defaults_from_attrs("name")
+    def run(self, name: str = None):
         """
         Returns the value of an environment variable after applying an optional `cast` function.
 
+        Args:
+            - name (str, optional): the name of the underlying environment variable to retrieve. Defaults
+                to the name provided at initialization.
+
         Returns:
             - Any: the (optionally type-cast) value of the environment variable
+
+        Raises:
+            - ValueError: if `raise_is_missing` is `True` and the environment variable was not found
         """
-        if self.raise_if_missing and self.env_var not in os.environ:
-            raise ValueError("Environment variable not set: {}".format(self.env_var))
-        value = os.getenv(self.env_var)
+        if self.raise_if_missing and name not in os.environ:
+            raise ValueError("Environment variable not set: {}".format(name))
+        value = os.getenv(name)
         if value is not None and self.cast is not None:
             value = self.cast(value)
         return value