[task scheduling] avoid relative module path problem with subcription keys

[zzstoatzz]

previously (as discussed here) we relied on the task_key as set by Task.__init__ to match task runs on TaskRuns to those a TaskServer was given at start-up, but this had a sneaky problem

say you setup your tasks and serve them:

from prefect import task
from prefect.task_server import serve

@task
def thing():
    print('hi')

if __name__ == '__main__':
    serve(thing)

Here: task_key == '__main__.thing'

but if you import and then submit that task somewhere else:

from myapp.tasks import thing

# ...

await thing.submit()

you end up with thing.task_key == 'myapp.tasks.thing'

so the TaskServer would never pick these up

---

This PR makes the subscription keys a hash of the task's name and the absolute path of where the task was defined, to ensure we don't have key mismatches related to relative module paths, without requiring a somewhat unintuitive thing like from . import tasks

[netlify]

✅ Deploy Preview for prefect-docs-preview ready!

Name	Link
🔨 Latest commit	d156203
🔍 Latest deploy log	https://app.netlify.com/sites/prefect-docs-preview/deploys/65f067ac7a95f90008693480
😎 Deploy Preview	https://deploy-preview-12227--prefect-docs-preview.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify site configuration.

[zzstoatzz]

the only way we should get here is if we cannot get the source file of the function directly - I'm not actually sure in what real world case that would be (i can make it happen by passing something like math.sqrt), but trying indiscriminately to get the source file we failed some of the test_tasks.py cases when a task is instantiated from a callable object like

class A:
    def __call__(self, *_args: Any, **_kwargs: Any) -> Any:
        return "hello"
        
task_instance = Task(fn=A)

even though I seem to have no problem submitting runs of this task instance 🤔

should be an extreme edge case but just flagging in case anyone has thoughts

[abrookins]

Interesting. @zzstoatzz Are you saying that the example of a callable object doesn't work?

[zzstoatzz]

it actually does!

even though I seem to have no problem submitting runs of this task instance 🤔

but I'm saying without the try/except in Task.__init__ which handles failures by setting this hash to None, we fail some of the test_tasks.py tests where we are testing things about tasks instantiated in this way

I tried to reproduce the failure I saw in the test by creating a task from this callable class A and then submitting it, but it actually worked. So it wasn't the most satisfying resolution but wanted to flag in case someone had a better idea of what was happening

[chrisguidry]

Hmm, won't this obscure the task key for all functions? I thought we only wanted to do this for ones that fell into the __main__ trap?

[zzstoatzz]

hmm yeah it does, updated in 7507ba3

[chrisguidry]

I dig it!

[discdiver]

With 2.16.4 the import is still required in the example below.

task_server.py:

from prefect import task
from prefect.task_server import serve


@task
def my_b_task(name: str):
    print(f"Hello, {name}!")
    return f"Hello, {name}!"


if __name__ == "__main__":
    from task_server import my_b_task  # this is required for the task to run 

    serve(my_b_task)

task_submitter.py

from task_server import my_b_task
import asyncio

if __name__ == "__main__":
    val = my_b_task.submit("Agrajag")
    print(val)

However, with a FastAPI server it is not if you just hit the API endpoint

first_fastapi.py

from fastapi import FastAPI
from prefect import task
from ff_prefect_task_server import my_fastapi_task

app = FastAPI()


@task
def my_b_task(name: str):
    print(f"Hello, {name}!")
    return f"Hello, {name}!"


@app.get("/ptask")
async def prefect_task():
    val = my_fastapi_task.submit(name="Marvin")
    return {"message": f"Prefect Task submitted: {val}"}

ff_prefect_task_server.py

from prefect import task
from prefect.task_server import serve


@task
def my_fastapi_task(name: str):
    print(f"Hello, {name}!")


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

[zzstoatzz]

hey @discdiver - thanks, I'll look at this today