https://prefect.io logo
Join SlackCommunities
Powered by
Is there anyway to stop a run if a task doesn't fi...
# ask-community
a
Is there anyway to stop a run if a task doesn't find any data but without it failing the run? I found the following code at https://discourse.prefect.io/t/how-can-i-stop-the-task-run-based-on-a-custom-logic/83 but I'd like it to just say "skipped" instead of it showing as a failure. 
from prefect import task
@task
def signal_task(message):
if message == 'stop_immediately!':
raise RuntimeError(message='Got a signal to end the task run!')
s
Hi, @Ash. Take a look at the docs on "state": https://docs.prefect.io/latest/concepts/states/#final-state-determination In particular, you can have your flow return a "COMPLETED" state if you like. Another approach is to simply have your flow/tasks handle the case where no data is found. I have some API scraping code that scrapes over date ranges. If a date range has nothing in it, I simply move on to the next date range and do not write a file for the empty date range.
upvote 1
a
Thanks @Sean Davis, I'll dig into this. Would it be possible to post a snippet of how you've utilised this?
s
In pseudocode, handling the case that no results are found. The result of something like this will be log messages that differ, but the flow will have a "COMPLETED" result. Note that you don't have to resort to the details of the docs I mentioned above. Just write your python code as you see fit, but either use try/catch (which I didn't do here) or simply deal with the case of no data directly (which I did model here).
Copy code
@task
def get_data(): #probably provide parameters here....
    logger = get_run_logger()
    results = ... # do your data fetching logic
    if len(results)==0:
        logger.info('no results found')
        return None
    logger.info('{len(results)} found')
    return results

@flow
def main_flow():
    data_results = get_data()
    if data_results is None:
        logger.info('Just repeating that no results found, but no worries, this is expected')
    elif:
        logger.info('Again, just repeating that {len(results)} found')
a
That's great, I'm on the right track then. I didn't like the idea of having multiple if else statements to check the results of various tasks (was hoping could just stop the flow from within the task) to then determine the state of the flow but guess that's how it's done! Appreciate the help :)
I'm still learning Python too so probably a better way of doing it but this seems to be working
Copy code
maximum_question_id = extract_maximum_question_id(questions_table_name, connection_string)

    if maximum_question_id == "No data in table.":
        slack_message(server + "." + schema + ": Table appears to be empty, no max ID found.")
        return Failed(message="Table appears to be empty, no max ID found.")
    else:
        questions_data = extract_questions(url, maximum_question_id)
        
        if questions_data[0] == "Can't reach URL":
            slack_message(server + "." + schema + ": Can't reach URL - status code " + str(questions_data[1]))
            return Failed(message="Can't reach URL - status code " + str(questions_data[1]))            
        elif questions_data == "No data in response":
            Warning = State(type=StateType.CANCELLED, name="SKIPPED", message="JSON response is empty.")
            return Warning
        else:
            extracted_data = structure_questions(questions_data)
        
            questionnaire_dataframe = transform_questions(extracted_data)

            final = load_questions(connection_string, questions_table_name, questionnaire_dataframe)
            if final:
                if final[0] == "Issue when loading data":
                    slack_message(server + "." + schema + ": Issue when loading data - " + str(final[1]))
                    return Failed(message="Issue when loading data - " + str(final[1]))
s
It looks like you have a good grasp of the "State" docs, though, so you'll be prepared for whatever you need. Personally, I wouldn't consider "no new data" as a failure or as a "CANCELLED" run; I consider anything other than "COMPLETED" as something that I need to investigate, but the behavior of your flow is totally up to you.
a
I decided on using Cancelled as it makes it grey in the UI, but I make it display SKIPPED instead of cancelled. Just helped clarify the running of the flows when looking at the dashboard
🦜 1
42 Views
Open in Slack
PreviousNext
Powered by