https://prefect.io logo
Title
c

Constantino Schillebeeckx

03/11/2022, 3:45 PM
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: 🧵
a

Anna Geller

03/11/2022, 3:51 PM
Can you move the traceback to the thread?
c

Constantino Schillebeeckx

03/11/2022, 3:53 PM
sorry, I'm not sure what you're asking exaclty
interestingly enough, when I run this locally the
StartFlowRun
call works just fine
k

Kevin Kho

03/11/2022, 3:57 PM
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?
:upvote: 1
c

Constantino Schillebeeckx

03/11/2022, 4:00 PM
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.
k

Kevin Kho

03/11/2022, 4:01 PM
Thanks
c

Constantino Schillebeeckx

03/11/2022, 4:01 PM
Prefect Core Version: 0.15.13
k

Kevin Kho

03/11/2022, 4:02 PM
I think this might have to do with an old API token on the agent?
Oh no I think it’s because artifacts specifically use the task run id. One sec
Yes I think this is because artifacts need to be attached to a task. You have an artifact in the subflow?
c

Constantino Schillebeeckx

03/11/2022, 4:06 PM
just had to look up what an artifact is 🙂 - nope, no artifacts here
k

Kevin Kho

03/11/2022, 4:12 PM
Looks like
StartFlowRun
itself uses a link artifact. Can you use
create_flow_run
instead?
c

Constantino Schillebeeckx

03/11/2022, 5:17 PM
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()
k

Kevin Kho

03/11/2022, 5:19 PM
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(....)
c

Constantino Schillebeeckx

03/11/2022, 5:30 PM
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?
k

Kevin Kho

03/11/2022, 5:34 PM
If you define with
@task
, it’s just
.run()
. If you use the class, there is init and run