Hi All, I have a flow that will run in DEV,QA and PROD. the flow connects to a database and I saved the credentials info as secrets in Prefect cloud. The secret variable name has surffix for environments. For example: "username_DEV","username_QA" etc. In flow, I use this code to get the password and username: pasword_var = "password_$FLOW_ENV"     snowflake_pass = PrefectSecret(pasword_var)     user_var = "user_$FLOW_ENV"     snowflake_user = PrefectSecret(user_var) and the environment variable is configured in runconfig: runconfig = DockerRun(image="[imagename]",env={"FLOW_ENV": "TEST"}) It returns error that PrefectSecret can't find secret with name "password_$FLOW_ENV" questions: 1: Is this a good approach to get these sensitive data through prefect secrets? 2: I though $[environment variable name] gives me the environment variable value? 3: It seems that PrefectSecret running as a task and doesn't evaluate the parameter value in running time. Is that right?

if you are using f-strings, the right syntax would be:

FLOW_ENV = os.environ.get("FLOW_ENV")
password_var = f"password_{FLOW_ENV}"

Yah, tried. It doesn't work. I also tried to create a task just for getting the environment variable value and return it. and then use this code: env = get_env() pasword_var = f"password_{env}" snowflake_pass = PrefectSecret(pasword_var) user_var = f"user__{env}" snowflake_user = PrefectSecret(user_var) no lucky.... I will try hadcord the names and then use switch but it's not neat. oh this is the task code: @task def *get_env*():     logger = prefect.context.get("logger")     env = os.environ.get("FLOW_ENV")     logger.info(env) return env the get_env() runs well. It returns the value properly.

Here is how you can do this:

import os

from prefect.tasks.secrets import PrefectSecret
import prefect
from prefect import Flow, task


class EnvSecret(PrefectSecret):
    def __init__(self, name=None, **kwargs):
        self.environment = os.environ.get("FLOW_ENV")
        _name = f"{name}_{self.environment.upper()}"
        super().__init__(name=_name, **kwargs)


env_secret = EnvSecret("A_PASSWORD")


@task
def print_result(secret):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(secret)


with Flow("flow_sample") as flow:
    secret = env_secret(task_args={"name": "A_PASSWORD"})
    print_result(secret)

Hi Anna, Thanks for your reply. It still doesn't work... I found I need to move this code os.environ.get("FLOW_ENV") into a run method: class *EnvSecret*(SecretBase):         def *__init__*(self,  name=None, env_name=None, **kwargs):         self.secret_name = name         self.env_name = env_name         super().*__init__*(name=name, **kwargs)     def *run*(self, name: str = None):       _#  env_value = os.getenv(self.env_name)_         env_value = os.getenv("FLOW_ENV")         *print*(self.env_name )         if name is None:             name = self.secret_name         if name is None:             raise ValueError("A secret name must be provided.")         f_name = f"{name}_{env_value}"         *print*(f_name )         return _Secret(f_name).get() env_secret = EnvSecret("snowflake_password") env_secret2 = EnvSecret("snowflake_user") It works well with DaskExecutor.

Can you share full code? As long as env variable is set, the above should work

sure. I think the issue is caused by using DaskExecutor.... class EnvSecret(PrefectSecret): def __init__(self, name=None, **kwargs): self.environment = os.environ.get("FLOW_ENV") _name = f"{name}_{self.environment.upper()}" super().__init__(name=_name, **kwargs) env_secret = EnvSecret("A_PASSWORD") DBT_PROJECT="xxxxxx" @task def print_result(secret): logger = prefect.context.get("logger") logger.info(secret) Storage=Docker(dockerfile="Dockerfile", image_name="xxxxx", image_tag="latest") runconfig = DockerRun(image="xxxxx",env={"FLOW_ENV": "TEST"}) with Flow(FLOW_NAME, storage=Storage, executor=executors.LocalDaskExecutor(), run_config=runconfig) as flow: secret = env_secret(task_args={"name": "A_PASSWORD"}) print_result(secret)