Does anybody know of a way to get the input parameters of a task run to show up in the UI? Kind of like how

@task(result=PrefectResult())

makes the output of the task run appear in the UI, except I'm looking to do this with the inputs as well.

There is no functionality for that. What are you trying to do? Debug a failed task by knowing the inputs?

I'm trying to determine if Prefect is the right tool for a REST-heavy workflow. Each flow takes a single website domain as input, then each task does something with a different API. For example, one might look up the MX records for that domain, some might do A record lookups, and then for each IP found we might do further DNS lookups and so forth. We want the tool we use to orchestrate all of this to allow us to inspect a single flow run (a domain name) and see what the inputs and output of each REST API step were. The

PrefectResult

gives us the output we'd like, but I'm struggling to show the associated input in the UI as well.

I’ll ask the team more tomorrow but off the top of my head, we don’t have a concept of opting in to record task inputs. The only immediate thing I can think of that can help here is logging or notifications containing those inputs.

Yup, I thought the same way and tried it in logging, but the UI is much nicer. I'm just seeing Orion for the first time now and was hoping maybe it has this feature, but it doesn't sound like it. I appreciate the quick response and look forward to hearing back if the team thinks of some way to expose this functionality. Thanks!

Yes feel free to ping me if you don’t hear back

You can enter those inputs as Prefect Parameters and they will be displayed in the flow run page, here is an example Alternatively, you could name your task runs based on inputs

@Anna Geller I would love example code to enter inputs as Prefect Parameters. Is it just something like

@task
def my_adder_step(a, b):
    Parameter(a)
    Parameter(b)
    result = a + b
    return result

?

sure, here is an example:

from prefect import task, Flow, Parameter


@task(log_stdout=True)
def plus_one(x):
    print(x + 1)


with Flow("parametrized_flow") as flow:
    param = Parameter("x", default=1)
    plus_one(x=param)

Ahh, I see. What if the input comes from a previous step, though? Something like:

@task
def get_urls():
    response = requests.get("<http://homepage.example.com>")
    urls = response.json()["urls"]
    return urls

@task
def classify_webpage(url): # <-- `url` is the thing we want in the UI
    response = requests.get(url)
    return _is_spam(response.text)

with Flow("spam_detector") as flow:
    returned_urls = get_urls()
    classify_webpages.map(returned_urls)

The idea is we don't know what URLs are going to come back from the first task, but for each downstream task run we want the name of the URL in the UI. FWIW, the named task run seems promising but I wonder if we can use your idea to do it with `Parameter`s.

You could do it mapping over flows probably - each flow run would be for a different URL, but your get_urls() would need to return a list of dictionaries e.g. [dict(parameter_name=url)]:

from prefect import Flow, unmapped
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
from prefect.executors import LocalDaskExecutor


with Flow("MasterFlow_Mapped", executor=LocalDaskExecutor()) as flow:
    returned_urls = get_urls()
    mapped_flow_run_ids = create_flow_run.map(
        parameters=returned_urls,
        flow_name=unmapped("flow_name"),
        project_name=unmapped("project"),
    )

Ohhh, got it. So one flow kicks off multiple flows and we use parameters there. Thanks! I'll try the named task run idea first and then fall back onto this if it doesn't work. I appreciate your responsiveness!