ex3.py

ex3.py

import random
from prefect import flow, task    # NEW **** 

@task   # NEW ****
def call_unreliable_api():
    choices = [{"data": 42}, "failure"]
    res = random.choice(choices)
    if res == "failure":
        raise Exception("Our unreliable service failed")
    else:
        return res

@task   # NEW ****
def augment_data(data: dict, msg: str):
    data["message"] = msg
    return data

@task   # NEW ****
def write_results_to_database(data: dict):
    print(f"Wrote {data} to database successfully!")
    return "Success!"

@flow 
def pipeline3(msg: str):
    api_result = call_unreliable_api()
    augmented_data = augment_data(data=api_result, msg=msg)
    write_results_to_database(augmented_data)