For context I am using prefect, which is a workflow orchestration engine that uses function decorates to tag functions for use. Their approach is very similar to dask in the creation of work flows.

Say I want to apply a decorator (in this case a prefect.task) to a function defined out of scope, say from an external module, I can do something like:

from prefect import task, Task
from some_module import foo

# task is the decorate, and returns an instance of Task

task_foo = task(foo)

@task
def task_bar(a: int, b: float) -> float:
    return a * b

in the case of task_bar my environment in vscode is able to resolve the function parameters to (a: int, b: float) -> float.

In the case of task_foo is it not: (...) -> None.

Is there a way to apply some typing.ParamSpec to appropriately provide the type inference?

import typing

P = typing.ParamSpec('P', bound=foo)
task_foo: Task[P] = task(foo)

I realise that ParamSpec(..., bound=foo) is not the answer here, but I am hoping it highlights the intent here.

I have not been able to spot the right thing to do here. I have found:

Which is getting very close to the problem I am describing. Playing around with ParamSpec I am falling just a little short.

Edit:

Alright, not I think I am misunderstanding something, though this does not surprise me. Doing something as simple as:

from typing import ParamSpec, TypeVar
from prefect import task, Task
from some_module import foo

P = ParamSpec('P')
R = TypeVar('R')


task_foo: Task[P,R] = task(foo)

behaves exactly I want it to. But I am left very confused where the information is bubbled up from. The Task class does have similar type P R parameters attached to its methods. Is it just a case that the types are being 'returned' and captured into the type variables P and R?

Hope this makes sense!

For context I am using prefect, which is a workflow orchestration engine that uses function decorates to tag functions for use. Their approach is very similar to dask in the creation of work flows.

Say I want to apply a decorator (in this case a prefect.task) to a function defined out of scope, say from an external module, I can do something like:

from prefect import task, Task
from some_module import foo

# task is the decorate, and returns an instance of Task

task_foo = task(foo)

@task
def task_bar(a: int, b: float) -> float:
    return a * b

in the case of task_bar my environment in vscode is able to resolve the function parameters to (a: int, b: float) -> float.

In the case of task_foo is it not: (...) -> None.

Is there a way to apply some typing.ParamSpec to appropriately provide the type inference?

import typing

P = typing.ParamSpec('P', bound=foo)
task_foo: Task[P] = task(foo)

I realise that ParamSpec(..., bound=foo) is not the answer here, but I am hoping it highlights the intent here.

I have not been able to spot the right thing to do here. I have found:

https://github/python/typing/issues/1027

Which is getting very close to the problem I am describing. Playing around with ParamSpec I am falling just a little short.

Edit:

Alright, not I think I am misunderstanding something, though this does not surprise me. Doing something as simple as:

from typing import ParamSpec, TypeVar
from prefect import task, Task
from some_module import foo

P = ParamSpec('P')
R = TypeVar('R')


task_foo: Task[P,R] = task(foo)

behaves exactly I want it to. But I am left very confused where the information is bubbled up from. The Task class does have similar type P R parameters attached to its methods. Is it just a case that the types are being 'returned' and captured into the type variables P and R?

Hope this makes sense!

• python
• python-typing
• prefect

• Is some_module.foo type-annotated? – dROOOze Commented Jan 30 at 6:35
• Yes, it is type-annotated – user29426864 Commented Jan 30 at 8:00

1 Answer 1

There are plenty ways how to achieve this:

1.

from typing import ParamSpec, TypeVar
from prefect import task, Task
from some_module import foo

P = ParamSpec('P')
R = TypeVar('R')

task_foo: Task[P, R] = task(foo)

from typing import Callable, ParamSpec, TypeVar
from prefect import task, Task

P = ParamSpec("P")
R = TypeVar("R")

def typed_task(fn: Callable[P, R], **task_kwargs) -> Task[P, R]:
    return task(fn, **task_kwargs)

from some_module import foo
task_foo = typed_task(foo)

reveal_type(task_foo)

from typing import Protocol, ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")

class TaskLike(Protocol[P, R]):
    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R: ...
    def submit(self, *args: P.args, **kwargs: P.kwargs) -> R: ...

def typed_task(fn: Callable[P, R]) -> TaskLike[P, R]:
    return task(fn)

from typing import cast

task_foo = cast(Task[(int, str), float], task(foo))

from typing import ParamSpec, TypeVar
from prefect import task, Task
from some_module import foo

P = ParamSpec('P')
R = TypeVar('R')

task_foo: Task[P, R] = task(foo)