Hi all! Having bit of trouble with PrefectSecret running the task. 1. I have added JSON file containing ['api_key'] in Prefect Cloud > Team > Secrets 2. Created my custom Task class called 'saveChartmogulActivities'

class saveChartmogulActivities(Task):
    def __init__(self, apikey, **kwargs):
        self.apikey = apikey
        self.endpoint = ENDPOINT_CHARTMOGUL_ACTIVITIES_EXPORT
        self.tmp_file_path = TMP_FILE_PATH
        super().__init__(**kwargs)

    def run(self):
        return self.apikey

apikey seems to be passed as Task not as actual value. Flow looks like this, it seems to be something with my custom defined Task class, but could not figure it out or find quickly from community here. Could anybody tell just by looking at it? :)

with Flow('chartmogul-extract', run_config=RUN_CONFIG) as flow:

# Reads Secrets from Prefect Cloud > Team > Secrets
    chartmogul_secrets = PrefectSecret("CHARTMOGUL")
    chartmogul_apikey = chartmogul_secrets['api_key']

    # Extract
    cm_activities_task = saveChartmogulActivities(chartmogul_apikey, log_stdout=True)
    cm_activities_run = cm_activities_task()

Although when I pass apikey as hardcoded string value, it works.

Secret

tasks retrieve secrets during flow run, not while you are initializing your tasks and building your flows. You should retrieve any secrets you want with

Secret

tasks, and pass them to other task via their

run

method, not

__init__

I'll give it a shot. But its important that

cm_activities_task = saveChartmogulActivities(log_stdout=True)(apikey=chartmogul_apikey)
    cm_activities_run = cm_activities_task()

Or could it also be like this below?

cm_activities_task = saveChartmogulActivities(log_stdout=True)
cm_activities_run = cm_activities_task(apikey=chartmogul_apikey)

oops, turns out I messed up my solution. The former option is wrong, calls three times in total. My bad. The latter option is perfectly valid. Edit: fixed my example