Hi team, thanks a lot for Prefect, having a lot of fun working with it so far! I have two questions 1. When mapping over results of a previous task, is it possible to only continue with the ones that were successful? In my use case, I first fetch feeds for a list of urls with

fetch_feed_for_url

and then save them with

save_feed_entries

(with trigger

any_successful

). However, some of the urls might not work leading to all subsequent tasks mapping over the result (and with that the flow) also failing. What I'm looking for is to map only over "successful

feeds_list

entries"

feeds_list = fetch_feed_for_url.map(url=feed_urls)
    save_feed_entries = save_feed_entries_to_db.map(feed=feeds_list)

2. Is it possible to initialize a task from a Task class using

EnvVarSecret

? For example, creating a Task using the Task Library often requires the credentials upfront. Are all the secrets required to be available at "flow building time" to work with Task classes?

fetch_mysql = MySQLFetch(
  user=EnvVarSecret("USER)
  password=EnvVarSecret("PASSWORD)
)

Thanks!

Hey @Jonas Bernhard, Glad you're having fun with Prefect! To your questions: 1. By default, mapped tasks will have a successful trigger by default so the downstream tasks on that mapped child won't kick off due to the mapped task ending in failure. If however the task succeeds, but your expected data is incorrect, then we may need some custom logic there. 2. Secrets are not required at build time, but rather runtime however you may need to instantiate your

EnvVarSecret

outside of your

fetch_mysql

task because ultimately secrets are also tasks and we need the value of a successful taskrun.

1. Thanks, I thought the

any_successful

behavior was what I was looking for, but in fact removing it solved the problem 👍 2. Sorry, I did not quite get how I can instantiate the

EnvVarSecret

in a way that can be used by task classes Running

class SayHiToUser(Task):
    def __init__(self, username, **kwargs):
        self.username = username
        super().__init__(**kwargs)

    def run(self):
        print(f"Hi {self.username}")


with Flow("EnvVar Test") as flow:
    username = EnvVarSecret("NAME")
    say_hi = SayHiToUser(username=username)
    hi = say_hi()

flow.run()

prints

"Hi <Task: NAME>"

since the class is instantiated not with the result of the

EnvVarSecret

task but with the task itself

I think in this case, you'll need to reconfigure your run method because you can't instantiate a task with values you do not know. Ultimately, you'll never know the value of

EnvVarSecret

when instantiating your task because that value is always deferred to runtime.

Thanks, that is the way we used it so far. Was just wondering if there was a way to use the tasks from the Task Library with env secrets but then we'll have to work around that for now 👍