`prefect.input.run_input`

This module contains functions that allow sending type-checked RunInput data to flows at runtime. Flows can send back responses, establishing two-way channels with senders. These functions are particularly useful for systems that require ongoing data transfer or need to react to input quickly. real-time interaction and efficient data handling. It’s designed to facilitate dynamic communication within distributed or microservices-oriented systems, making it ideal for scenarios requiring continuous data synchronization and processing. It’s particularly useful for systems that require ongoing data input and output. The following is an example of two flows. One sends a random number to the other and waits for a response. The other receives the number, squares it, and sends the result back. The sender flow then prints the result. Sender flow:

import random
from uuid import UUID
from prefect import flow
from prefect.logging import get_run_logger
from prefect.input import RunInput

class NumberData(RunInput):
    number: int


@flow
async def sender_flow(receiver_flow_run_id: UUID):
    logger = get_run_logger()

    the_number = random.randint(1, 100)

    await NumberData(number=the_number).send_to(receiver_flow_run_id)

    receiver = NumberData.receive(flow_run_id=receiver_flow_run_id)
    squared = await receiver.next()

    logger.info(f"{the_number} squared is {squared.number}")

Receiver flow:

import random
from uuid import UUID
from prefect import flow
from prefect.logging import get_run_logger
from prefect.input import RunInput

class NumberData(RunInput):
    number: int


@flow
async def receiver_flow():
    async for data in NumberData.receive():
        squared = data.number ** 2
        data.respond(NumberData(number=squared))

Functions

`keyset_from_paused_state`

keyset_from_paused_state(state: 'State') -> Keyset

Get the keyset for the given Paused state. Args:

- state: the state to get the keyset for

`keyset_from_base_key`

keyset_from_base_key(base_key: str) -> Keyset

Get the keyset for the given base key. Args:

- base_key: the base key to get the keyset for

Returns:

- Dict[str, str]: the keyset

`run_input_subclass_from_type`

run_input_subclass_from_type(_type: Union[Type[R], Type[T], pydantic.BaseModel]) -> Union[Type[AutomaticRunInput[T]], Type[R]]

Create a new RunInput subclass from the given type.

`asend_input`

asend_input(run_input: Any, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Send input to a flow run asynchronously.

`send_input`

send_input(run_input: Any, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Send input to a flow run.

`receive_input`

receive_input(input_type: Union[Type[R], Type[T], pydantic.BaseModel], timeout: Optional[float] = 3600, poll_interval: float = 10, raise_timeout_error: bool = False, exclude_keys: Optional[Set[str]] = None, key_prefix: Optional[str] = None, flow_run_id: Optional[UUID] = None, with_metadata: bool = False) -> Union[GetAutomaticInputHandler[T], GetInputHandler[R]]

Classes

`RunInputMetadata`

`BaseRunInput`

Methods:

`aload`

aload(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> Self

Load the run input response from the given key asynchronously. Args:

- keyset: the keyset to load the input for
- flow_run_id: the flow run ID to load the input for

`arespond`

arespond(self, run_input: 'RunInput', sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Respond to the sender of this input asynchronously.

`asave`

asave(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> None

Save the run input response to the given key asynchronously. Args:

- keyset: the keyset to save the input for
- flow_run_id: the flow run ID to save the input for

`asend_to`

asend_to(self, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Send this input to a flow run asynchronously.

`keyset_from_type`

keyset_from_type(cls) -> Keyset

`load`

load(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> Self

Load the run input response from the given key. Args:

- keyset: the keyset to load the input for
- flow_run_id: the flow run ID to load the input for

`load_from_flow_run_input`

load_from_flow_run_input(cls, flow_run_input: 'FlowRunInput') -> Self

Load the run input from a FlowRunInput object. Args:

- flow_run_input: the flow run input to load the input for

`metadata`

metadata(self) -> RunInputMetadata

`respond`

respond(self, run_input: 'RunInput', sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Respond to the sender of this input.

`save`

save(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> None

Save the run input response to the given key. Args:

- keyset: the keyset to save the input for
- flow_run_id: the flow run ID to save the input for

`send_to`

send_to(self, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Send this input to a flow run.

`with_initial_data`

with_initial_data(cls: Type[R], description: Optional[str] = None, **kwargs: Any) -> Type[R]

Create a new RunInput subclass with the given initial data as field defaults. Args:

- description: a description to show when resuming a flow run that requires input
- kwargs: the initial data to populate the subclass

`RunInput`

Methods:

`aload`

aload(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> Self

Load the run input response from the given key asynchronously. Args:

- keyset: the keyset to load the input for
- flow_run_id: the flow run ID to load the input for

`arespond`

arespond(self, run_input: 'RunInput', sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Respond to the sender of this input asynchronously.

`asave`

asave(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> None

Save the run input response to the given key asynchronously. Args:

- keyset: the keyset to save the input for
- flow_run_id: the flow run ID to save the input for

`asend_to`

asend_to(self, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Send this input to a flow run asynchronously.

`keyset_from_type`

keyset_from_type(cls) -> Keyset

`load`

load(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> Self

Load the run input response from the given key. Args:

- keyset: the keyset to load the input for
- flow_run_id: the flow run ID to load the input for

`load_from_flow_run_input`

load_from_flow_run_input(cls, flow_run_input: 'FlowRunInput') -> Self

Load the run input from a FlowRunInput object. Args:

- flow_run_input: the flow run input to load the input for

`metadata`

metadata(self) -> RunInputMetadata

`receive`

receive(cls, timeout: Optional[float] = 3600, poll_interval: float = 10, raise_timeout_error: bool = False, exclude_keys: Optional[Set[str]] = None, key_prefix: Optional[str] = None, flow_run_id: Optional[UUID] = None) -> GetInputHandler[Self]

`respond`

respond(self, run_input: 'RunInput', sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Respond to the sender of this input.

`save`

save(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> None

Save the run input response to the given key. Args:

- keyset: the keyset to save the input for
- flow_run_id: the flow run ID to save the input for

`send_to`

send_to(self, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Send this input to a flow run.

`subclass_from_base_model_type`

subclass_from_base_model_type(cls, model_cls: Type[pydantic.BaseModel]) -> Type['RunInput']

Create a new RunInput subclass from the given pydantic.BaseModel subclass. Args:

- model_cls: the class from which to create the new RunInput subclass

`with_initial_data`

with_initial_data(cls: Type[R], description: Optional[str] = None, **kwargs: Any) -> Type[R]

Create a new RunInput subclass with the given initial data as field defaults. Args:

- description: a description to show when resuming a flow run that requires input
- kwargs: the initial data to populate the subclass

`AutomaticRunInput`

Methods:

`aload`

aload(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> T

Load the run input response from the given key asynchronously. Args:

- keyset: the keyset to load the input for
- flow_run_id: the flow run ID to load the input for

`aload`

aload(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> Self

Load the run input response from the given key asynchronously. Args:

- keyset: the keyset to load the input for
- flow_run_id: the flow run ID to load the input for

`arespond`

arespond(self, run_input: 'RunInput', sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Respond to the sender of this input asynchronously.

`asave`

asave(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> None

Save the run input response to the given key asynchronously. Args:

- keyset: the keyset to save the input for
- flow_run_id: the flow run ID to save the input for

`asend_to`

asend_to(self, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Send this input to a flow run asynchronously.

`keyset_from_type`

keyset_from_type(cls) -> Keyset

`load`

load(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> T

Load the run input response from the given key. Args:

- keyset: the keyset to load the input for
- flow_run_id: the flow run ID to load the input for

`load`

load(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> Self

Load the run input response from the given key. Args:

- keyset: the keyset to load the input for
- flow_run_id: the flow run ID to load the input for

`load_from_flow_run_input`

load_from_flow_run_input(cls, flow_run_input: 'FlowRunInput') -> Self

Load the run input from a FlowRunInput object. Args:

- flow_run_input: the flow run input to load the input for

`metadata`

metadata(self) -> RunInputMetadata

`receive`

receive(cls, timeout: Optional[float] = 3600, poll_interval: float = 10, raise_timeout_error: bool = False, exclude_keys: Optional[Set[str]] = None, key_prefix: Optional[str] = None, flow_run_id: Optional[UUID] = None, with_metadata: bool = False) -> GetAutomaticInputHandler[T]

`respond`

respond(self, run_input: 'RunInput', sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Respond to the sender of this input.

`save`

save(cls, keyset: Keyset, flow_run_id: Optional[UUID] = None) -> None

Save the run input response to the given key. Args:

- keyset: the keyset to save the input for
- flow_run_id: the flow run ID to save the input for

`send_to`

send_to(self, flow_run_id: UUID, sender: Optional[str] = None, key_prefix: Optional[str] = None) -> None

Send this input to a flow run.

`subclass_from_type`

subclass_from_type(cls, _type: Type[T]) -> Type['AutomaticRunInput[T]']

Create a new AutomaticRunInput subclass from the given type. This method uses the type’s name as a key prefix to identify related flow run inputs. This helps in ensuring that values saved under a type (like List[int]) are retrievable under the generic type name (like “list”).

`with_initial_data`

with_initial_data(cls: Type[R], description: Optional[str] = None, **kwargs: Any) -> Type[R]

Create a new RunInput subclass with the given initial data as field defaults. Args:

- description: a description to show when resuming a flow run that requires input
- kwargs: the initial data to populate the subclass

`GetInputHandler`

Methods:

`anext`

anext(self) -> R

Get the next input asynchronously.

`filter_for_inputs`

filter_for_inputs(self) -> list['FlowRunInput']

Filter for inputs asynchronously.

`filter_for_inputs_sync`

filter_for_inputs_sync(self) -> list['FlowRunInput']

Filter for inputs synchronously.

`next`

next(self) -> R

Get the next input.

`to_instance`

to_instance(self, flow_run_input: 'FlowRunInput') -> R

`GetAutomaticInputHandler`

Methods:

`anext`

anext(self) -> Union[T, AutomaticRunInput[T]]

Get the next input asynchronously.

`filter_for_inputs`

filter_for_inputs(self) -> list['FlowRunInput']

Filter for inputs asynchronously.

`filter_for_inputs_sync`

filter_for_inputs_sync(self) -> list['FlowRunInput']

Filter for inputs synchronously.

`next`

next(self) -> Union[T, AutomaticRunInput[T]]

Get the next input.

`to_instance`

to_instance(self, flow_run_input: 'FlowRunInput') -> Union[T, AutomaticRunInput[T]]

​prefect.input.run_input

​Functions

​keyset_from_paused_state

​keyset_from_base_key

​run_input_subclass_from_type

​asend_input

​send_input

​receive_input

​Classes

​RunInputMetadata

​BaseRunInput

​aload

​arespond

​asave

​asend_to

​keyset_from_type

​load

​load_from_flow_run_input

​metadata

​respond

​save

​send_to

​with_initial_data

​RunInput

​aload

​arespond

​asave

​asend_to

​keyset_from_type

​load

​load_from_flow_run_input

​metadata

​receive

​respond

​save

​send_to

​subclass_from_base_model_type

​with_initial_data

​AutomaticRunInput

​aload

​aload

​arespond

​asave

​asend_to

​keyset_from_type

​load

​load

​load_from_flow_run_input

​metadata

​receive

​respond

​save

​send_to

​subclass_from_type

​with_initial_data

​GetInputHandler

​anext

​filter_for_inputs

​filter_for_inputs_sync

​next

​to_instance

​GetAutomaticInputHandler

​anext

​filter_for_inputs

​filter_for_inputs_sync

​next

​to_instance

`prefect.input.run_input`

Functions

`keyset_from_paused_state`

`keyset_from_base_key`

`run_input_subclass_from_type`

`asend_input`

`send_input`

`receive_input`

Classes

`RunInputMetadata`

`BaseRunInput`

`aload`

`arespond`

`asave`

`asend_to`

`keyset_from_type`

`load`

`load_from_flow_run_input`

`metadata`

`respond`

`save`

`send_to`

`with_initial_data`

`RunInput`

`aload`

`arespond`

`asave`

`asend_to`

`keyset_from_type`

`load`

`load_from_flow_run_input`

`metadata`

`receive`

`respond`

`save`

`send_to`

`subclass_from_base_model_type`

`with_initial_data`

`AutomaticRunInput`

`aload`

`aload`

`arespond`

`asave`

`asend_to`

`keyset_from_type`

`load`

`load`

`load_from_flow_run_input`

`metadata`

`receive`

`respond`

`save`

`send_to`

`subclass_from_type`

`with_initial_data`

`GetInputHandler`

`anext`

`filter_for_inputs`

`filter_for_inputs_sync`

`next`

`to_instance`

`GetAutomaticInputHandler`

`anext`

`filter_for_inputs`

`filter_for_inputs_sync`

`next`

`to_instance`