My flow run may sometimes fail and I want to be able to restart it. Because of that I'm using the

flow.result

to store checkpoints. However, some tasks contain secret values that I don't want to store anywhere - so I've subclassed the

SecretBase

which indeed doesn't store the result. However, when I restart the flow from the failed state, the value of the secret is

None

(I suppose because there is no

Result

object for it), however I would expect it to be re-computed. Below is a simple example I was able to reproduce my issue with, here is the error I get when I re-run the flow from failed state. Is there something I'm doing wrong?

Unexpected error: TypeError("'NoneType' object is not subscriptable",)
Traceback (most recent call last):
  File ".../home/user/bss/dice/Tournesol/prefect_flows/flows//venv/lib/python3.6/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File ".../venv/lib/python3.6/site-packages/prefect/engine/task_runner.py", line 860, in get_task_run_state
    logger=self.logger,
  File ".../venv/lib/python3.6/site-packages/prefect/utilities/executors.py", line 298, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File ".../secret_result.py", line 35, in use_secret
    <http://prefect.context.logger.info|prefect.context.logger.info>(secret['path'])
TypeError: 'NoneType' object is not subscriptable

import os
import random
from os import environ

import prefect
from prefect import task, Flow, Parameter
from prefect.engine.results import LocalResult
from prefect.tasks.secrets import SecretBase

result_directory = os.path.abspath(
    os.path.join(os.path.dirname(__file__), "..", "results")
)

result = LocalResult(
    dir=result_directory,
    location="{flow_name}/"
    "{scheduled_start_time:%d-%m_%H-%M-%S}/"
    "{task_full_name}-{task_run_id}.prefect_result",
)


class CustomSecret(SecretBase):
    def run(self):
        return {'path': environ.get('PATH'), 'home': environ.get('HOME')}


@task
def fragile_task(error_probability: float) -> None:
    if random.random() <= error_probability:
        raise ValueError('I\'m fragile, just retry me')


@task
def use_secret(secret: CustomSecret) -> None:
    <http://prefect.context.logger.info|prefect.context.logger.info>(secret['path'])
    <http://prefect.context.logger.info|prefect.context.logger.info>(secret['home'])


with Flow('test', result=result) as flow:
    error_prob = Parameter('error_probability', default=0.7)
    secret = CustomSecret()

    a = fragile_task(error_prob)
    b = use_secret(secret)

    flow.set_dependencies(b, upstream_tasks=[a])

if __name__ == '__main__':
    flow.register(project_name='default')

@Marvin open “Recompute secret tasks when retrying flow from failed”

Thanks for opening the issue though 👍 will keep an eye on it