CODEWITHSUNDEEP
pythonpython-typingtemporal-workflow
spacether
spacether

Reputation: 2721

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

Upvotes: 0

Views: 71

Answers (1)

Jasmijn
Jasmijn

Reputation: 10477

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

Upvotes: 1

Related Questions