Typing decorators that turn *args, **kwargs into args, kwargs

[henribru]

We now have ParamSpec that easily lets you represent the exact signature of an arbitrary function. I'm however curious about a slightly different case, where you have a callable that takes the same arguments as another function, but as separate args and kwargs arguments that aren't unpacked. This is common in task queues, here are some examples:

https://docs.celeryq.dev/en/stable/reference/celery.app.task.html#celery.app.task.Task.apply
https://huey.readthedocs.io/en/latest/api.html#TaskWrapper.schedule
https://dramatiq.io/reference.html#dramatiq.Actor.send_with_options
https://python-rq.org/docs/

To elaborate on the first example, the way it works is:

from celery import Celery

app = Celery(...)

@app.task
def add(x, y):
    return x + y

add.apply(args=(1, 1))
# or
add.apply(kwargs={"x": 1, "y": 1})

Notice how it's not add.apply(1, 1), because it doesn't unpack the arguments. There is a similar method that does just that: https://docs.celeryq.dev/en/stable/reference/celery.app.task.html#celery.app.task.Task.delay. The latter is easy to type with ParamSpec. You'll just do something like

from typing import ParamSpec, Callable, Generic, Any


P = ParamSpec("P")

class Task(Generic[P]):
    def delay(self, *args: P.args, **kwargs: P.kwargs): ...

def task(f: Callable[P, Any]) -> Task[P]: ...

You can't type apply though, at least not in the most straight forward way of def apply(args: P.args, kwargs: P.kwargs): .... That's not a valid use location according to the ParamSpec PEP: https://peps.python.org/pep-0612/#id1

Is there another way to type this? If not, have there been considerations to expanding on the ParamSpec spec to support a case like this?

[hmc-cs-mdrissi]

For args case you could use typevartuple (pep 646) to describe it. Something like

from typing import TypeVar, Protocol, Generic
from typing_extensions import TypeVarTuple, Unpack

R = TypeVar('R')
Ts = TypeVarTuple('Ts')

class TaskFn(Protocol, Generic[R, Ts]):
  def apply(self, args: tuple[Unpack[Ts]]) -> R:
    ...

def task(func: Callable[[Unpack[Ts]], R]) -> TaskFn[R, Ts]:
  ...

I don't see a way to do anything for keyword args.

For paramspec currently P.args/P.kwargs are only allowed to be used with *args/**kwags. You can't use them elsewhere. It'd be a bit messy as what happens if you pass both?

add.apply(args=(1,), kwargs={"y": 1}) 
add.apply(args=(1,1), kwargs={"x": 1, "y": 1})

Also how would rules for things like required arguments be handled? Would signature be allowed to use P.args/P.kwargs multiple times if it was allowed for non *args/**kwargs?

[gandhis1]

You mind elaborating on what would be wrong with passing both - the possibility of a collision between a keyword argument and a positional argument? Wouldn't the type checker be able to identify that when the actual Callable is called?

[PeterJCLaw]

Not sure this is the best place to ask .. but .. supporting this use-case feels like it would be quite useful. Do we think there's a likelihood that support for this might appear at some point?

The background tasks use-case is what I'd had in mind and came looking for a discussion around how to type this. There's a Django proposal which is exploring adding a general interface for tasks (django/deps#86) and it would be great if that could be designed from the start to have typing support.