I'm working on writing a state handler that executes some logic at the very end of a flow, after all tasks have run. In that logic I end up calling

StartFlowRun(...).run()

- when this executes in the cloud I'm seeing: 🧵

Can you move the traceback to the thread?

sorry, I'm not sure what you're asking exaclty

Anna is asking if we could have the error message in this thread to keep the main channel cleaner. What Prefect version are you on for Flow and agent?

Exception raised while calling state handlers: ClientError('400 Client Error: Bad Request for url: <https://api.prefect.io/graphql>\n\nThe following error messages were provided by the GraphQL server:\n\n    INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at\n        "input.task_run_id"; Expected non-nullable type UUID! not to be null.\n\nThe GraphQL query was:\n\n    mutation($input: create_task_run_artifact_input!) {\n            create_task_run_artifact(input: $input) {\n                id\n        }\n    }\n\nThe passed variables were:\n\n    {"input": {"task_run_id": null, "kind": "link", "data": {"link": "/dh-data-team/flow-run/0d1b3659-c0f4-43c0-8b3f-6396f7311d82"}, "tenant_id": null}}\n')
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 648, in _send_request
    response.raise_for_status()
  File "/usr/local/lib/python3.9/site-packages/requests/models.py", line 960, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: <https://api.prefect.io/graphql>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/prefect/engine/cloud/flow_runner.py", line 119, in call_runner_target_handlers
    new_state = super().call_runner_target_handlers(
  File "/usr/local/lib/python3.9/site-packages/prefect/engine/flow_runner.py", line 116, in call_runner_target_handlers
    new_state = handler(self.flow, old_state, new_state) or new_state
  File "/usr/local/prefect/flows/utils/flow_deps.py", line 171, in check_flow_dep_handler
    deps.trigger_downtream_dependencies
  File "/usr/local/prefect/flows/utils/flow_deps.py", line 296, in trigger_downtream_dependencies
    StartFlowRun(
  File "/usr/local/lib/python3.9/site-packages/prefect/utilities/tasks.py", line 456, in method
    return run_method(self, *args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/prefect/tasks/prefect/flow_run.py", line 466, in run
    create_link_artifact(urlparse(run_link).path)
  File "/usr/local/lib/python3.9/site-packages/prefect/backend/artifacts.py", line 52, in create_link_artifact
    return _create_task_run_artifact("link", {"link": link})
  File "/usr/local/lib/python3.9/site-packages/prefect/backend/artifacts.py", line 28, in _create_task_run_artifact
    return client.create_task_run_artifact(
  File "/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 2160, in create_task_run_artifact
    result = self.graphql(
  File "/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 549, in graphql
    result = <http://self.post|self.post>(
  File "/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 452, in post
    response = self._request(
  File "/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 739, in _request
    response = self._send_request(
  File "/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 661, in _send_request
    raise ClientError(f"{exc}\n{graphql_msg}") from exc
prefect.exceptions.ClientError: 400 Client Error: Bad Request for url: <https://api.prefect.io/graphql>

The following error messages were provided by the GraphQL server:

    INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at
        "input.task_run_id"; Expected non-nullable type UUID! not to be null.

The GraphQL query was:

    mutation($input: create_task_run_artifact_input!) {
            create_task_run_artifact(input: $input) {
                id
        }
    }

The passed variables were:

    {"input": {"task_run_id": null, "kind": "link", "data": {"link": "/dh-data-team/flow-run/0d1b3659-c0f4-43c0-8b3f-6396f7311d82"}, "tenant_id": null}}

This makes me think I shouldn't be calling a

Task

from within the handler but I'm not sure.

Thanks

Prefect Core Version: 0.15.13

I think this might have to do with an old API token on the agent?

just had to look up what an artifact is 🙂 - nope, no artifacts here

Looks like

StartFlowRun

itself uses a link artifact. Can you use

create_flow_run

instead?

Moved over to use `create_flow_run`:

Exception raised while calling state handlers: ValueError("Could not infer an active Flow context while creating edge to <Task: create_flow_run>. This often means you called a task outside a `with Flow(...)` block. If you're trying to run this task outside of a Flow context, you need to call `create_flow_run.run(...)`")
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/prefect/engine/cloud/flow_runner.py", line 119, in call_runner_target_handlers
    new_state = super().call_runner_target_handlers(
  File "/usr/local/lib/python3.9/site-packages/prefect/engine/flow_runner.py", line 116, in call_runner_target_handlers
    new_state = handler(self.flow, old_state, new_state) or new_state
  File "/usr/local/prefect/flows/utils/flow_deps.py", line 170, in check_flow_dep_handler
    deps.trigger_downtream_dependencies
  File "/usr/local/prefect/flows/utils/flow_deps.py", line 300, in trigger_downtream_dependencies
    create_flow_run(
  File "/usr/local/lib/python3.9/site-packages/prefect/core/task.py", line 633, in __call__
    new.bind(
  File "/usr/local/lib/python3.9/site-packages/prefect/core/task.py", line 693, in bind
    raise ValueError(
ValueError: Could not infer an active Flow context while creating edge to <Task: create_flow_run>. This often means you called a task outside a `with Flow(...)` block. If you're trying to run this task outside of a Flow context, you need to call `create_flow_run.run(...)`

here's what the call looks like

create_flow_run(
    flow_name=flow,
    project_name=PREFECT_PROJECT,
    parameters=params,
    scheduled_start_time=sst,
).run()

there is no init-run separation for this one because it was created with the task decorator so move everything into the run:

create_flow_run.run(....)

that did it, beautiful. I don't quite understand this style of task invocation - are there some docs somewhere I can read up on this?

If you define with

@task

, it’s just

.run()

. If you use the class, there is init and run