prefect rest api, create task run fail

[shuzhiwei]

Bug summary

@router.post("/create_task_run")
async def create_task_run(job_id: str = Form(...)):
    """
    create_task_run
    """
    from workflow.prepare import create_workdir

    task_key = _generate_task_key(create_workdir)
    dynamic_key = f"{task_key}-{str(uuid4().hex)}"
    data = {
        "name": "create_workdir",
        "task_key": task_key,
        "dynamic_key": dynamic_key,
    }
    run_url = f"{PREFECT_BASE_URL}/api/task_runs/"
    res = requests.post(run_url, json.dumps(data))
    return res.json()

workflow.prepare is a flow, and create_workdir is a task，
Assume that functions such as _generate_task_key, uuid4, and so on already exist.
After the above code executes, the task is submitted and remains in Pending state！What is the reason? How could this task running normally？

Version info

Version:             3.0.3
API version:         0.8.4
Python version:      3.11.7
Git commit:          d9c51bc3
Built:               Fri, Sep 20, 2024 8:52 AM

Additional context

No response

[desertaxle]

Hey @shuzhiwei! It sounds like you're creating a task run via the REST API and expecting it to be executed, is that correct? Are you running a task worker for the task you're trying to execute? This page in the documentation has information about running background tasks with Prefect: https://docs.prefect.io/3.0/develop/deferred-tasks.

[shuzhiwei]

Hey @desertaxle! Thank you very much for your documentation.
However, I followed the documentation and found a problem:

task_worker.py

from prefect import task
from prefect.task_worker import serve


@task(persist_result=False,
      log_prints=True
      )
def my_background_task(name: str):
    print(f"Hello, {name}!")


if __name__ == "__main__":
    serve(my_background_task)

task_scheduler.py

from workflow.task_worker import my_background_task

if __name__ == "__main__":
    my_background_task.delay("Agrajag")

python3 workflow/task_worker.py execute is normal.

python3 workflow/task_scheduler.py raise prefect.exceptions.SerializationError: Failed to serialize object of type 'dict' with serializer 'json'. You can try a different serializer (e.g. result_serializer="json") or disabling persistence (persist_result=False) for this flow or task.

but I had add persist_result=False on task, Why is that?

[desertaxle]

The example code that your provided runs fine for me even without persist_result=False. Did you see that error on a different task? That error occurs when you return a non-serializable result from a task, so a value in your dictionary might not have been JSON serializable. You could try using the pickle serializer instead (which is the default), or you might need to change the return value of your task.

This thread seems like it's related to a usage question, so I'm going to convert this issue to a discussion, and we can keep chatting there.

[shuzhiwei]

Hey @desertaxle! I sincerely appreciate your response. After I used the pickle serializer，task_scheduler is worked！However，we hope to use json serializer, As in the previous example, once you change the serializer to json，task_scheduler will raise that error. Is this a bug? I eagerly await your response.

[menachescu]

Just noticed this bug and it seems the title of the issue should be updated to mention the json serialization issue