Search code examples
pythonpython-typingtemporal-workflow

Is it possible to type hint a callable that takes positional arguments only using a generic for the positional type?


In python, is it possible to type hint a callable that takes positional arguments only using a generic for the positional type?

The context is that I want a to ingest positional args only. See this temporalio code:

@overload
async def execute_activity(
    activity: Callable[..., Awaitable[ReturnType]],
    arg: None,
    *,
    args: Sequence[Any],
    task_queue: Optional[str] = None,
    schedule_to_close_timeout: Optional[timedelta] = None,
    schedule_to_start_timeout: Optional[timedelta] = None,
    start_to_close_timeout: Optional[timedelta] = None,
    heartbeat_timeout: Optional[timedelta] = None,
    retry_policy: Optional[temporalio.common.RetryPolicy] = None,
    cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
    activity_id: Optional[str] = None,
    versioning_intent: Optional[VersioningIntent] = None,
) -> ReturnType: ...

The doc is present here

The activity is a callable where any positional arguments are allowed. I want a generic for the ... input, so I can use it to pass in the typed arg input in the args input.

It looks like I could use ParamSpec but that allows keyword arguments too, and appears that it would not throw needed errors when keyword args are defined. Is there a way to do this in python where the input positional arg type in activity can be used as the args type in that signature? How can I do this?

My goal would be some psudocode like:

@overload
async def execute_activity(
    activity: Callable[GenericIterable, Awaitable[ReturnType]],
    arg: None,
    *,
    args: GenericIterable,
    task_queue: Optional[str] = None,
    schedule_to_close_timeout: Optional[timedelta] = None,
    schedule_to_start_timeout: Optional[timedelta] = None,
    start_to_close_timeout: Optional[timedelta] = None,
    heartbeat_timeout: Optional[timedelta] = None,
    retry_policy: Optional[temporalio.common.RetryPolicy] = None,
    cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
    activity_id: Optional[str] = None,
    versioning_intent: Optional[VersioningIntent] = None,
) -> ReturnType: ...

# where the below function could be input

def some_activity(a: int, b: str) -> float:
   # sample allowed activity definition
   return 3.14

The solution may be: don't do that, use one input and output type so that generics/type hints will do their job here.

Working answer is below

If you want this in temporal upvote this feature request: https://github.com/temporalio/sdk-python/issues/779


Solution

  • This might be a good use case for a TypeVarTuple.

    For example:

    # for Python 3.11
    from typing import TypeVarTuple
    
    Ts = TypeVarTuple("Ts")
    
    @overload
    async def execute_activity(
        activity: Callable[[*Ts], Awaitable[ReturnType]],
        *args: *Ts,
    # snip
    
    
    # for Python 3.12+
    @overload
    async def execute_activity[*Ts](
        activity: Callable[[*Ts], Awaitable[ReturnType]],
        *args: *Ts,
    # snip
    

    Note: for this to work, you must remove the arg parameter, and have *args cover the case where 1 arg is input or many args are input