https://prefect.io logo
Title
w

Wojciech Kieliszek

03/14/2023, 12:07 PM
Hi, in Prefect v1 I could raise a signal to set a task to a given state like this:
start_time = datetime.utcnow() + timedelta(seconds=retry_after)
        raise signals.RETRY(
            start_time=start_time,
            message=f"Rate limited. Waiting {retry_after}s.",
        )
This way I could implement my own retry logic (e.g. not based only on exception class, but exception data and define
retry_after
based on the API response). Is it possible with Prefect v2? I am searching the docs, but couldn’t find if returning or raising states manually is supported. Something like this:
start_time = pendulum.now("UTC").add(seconds=retry_after)
        return AwaitingRetry(
            scheduled_time=start_time,
            message=f"Rate limited. Waiting {retry_after}s.",
        )
1
r

Ryan Peden

03/14/2023, 12:31 PM
Hi Wojciech! In general, returning a state from a task should work in Prefect 2. In case you're interested, this is the section of code that converts task return values into states. As you can see, if you return a manually-created state, Prefect will use it. It looks like the docs mention this for flows, but not tasks - so I'll make a note to update the docs to make them more clear.
w

Wojciech Kieliszek

03/14/2023, 2:33 PM
Hi, thanks for the response. I tried returning a state from a task and it seems to work for Completed state, but when returning
AwaitingRetry
like:
start_time = pendulum.now("UTC").add(seconds=retry_after)
        return AwaitingRetry(
            scheduled_time=start_time,
            message=f"Rate limited. Waiting {retry_after}s.",
        )
I am getting:
This run cannot transition to the StateType.SCHEDULED state from the StateType.RUNNING state.
inside
set_task_run_state
. Does it mean we cannot manually retry task with custom
scheduled_time
?