prefect.input.actions

create_flow_run_input async

Create a new flow run input. The given value will be serialized to JSON and stored as a flow run input value.

Parameters:

Name	Type	Description	Default
-	key (str	the flow run input key	required
-	value (Any	the flow run input value	required
-	flow_run_id (UUID	the, optional, flow run ID. If not given will default to pulling the flow run ID from the current context.	required

Source code in prefect/input/actions.py

@sync_compatible
@inject_client
async def create_flow_run_input(
    key: str,
    value: Any,
    flow_run_id: Optional[UUID] = None,
    client: "PrefectClient" = None,
):
    """
    Create a new flow run input. The given `value` will be serialized to JSON
    and stored as a flow run input value.

    Args:
        - key (str): the flow run input key
        - value (Any): the flow run input value
        - flow_run_id (UUID): the, optional, flow run ID. If not given will
          default to pulling the flow run ID from the current context.
    """
    flow_run_id = _ensure_flow_run_id(flow_run_id)

    await client.create_flow_run_input(
        flow_run_id=flow_run_id, key=key, value=orjson.dumps(value).decode()
    )

delete_flow_run_input async

Delete a flow run input.

Parameters:

Name	Type	Description	Default
-	flow_run_id (UUID	the flow run ID	required
-	key (str	the flow run input key	required

Source code in prefect/input/actions.py

@sync_compatible
@inject_client
async def delete_flow_run_input(
    key: str, flow_run_id: Optional[UUID] = None, client: "PrefectClient" = None
):
    """Delete a flow run input.

    Args:
        - flow_run_id (UUID): the flow run ID
        - key (str): the flow run input key
    """

    flow_run_id = _ensure_flow_run_id(flow_run_id)

    await client.delete_flow_run_input(flow_run_id=flow_run_id, key=key)

read_flow_run_input async

Read a flow run input.

Parameters:

Name	Type	Description	Default
-	key (str	the flow run input key	required
-	flow_run_id (UUID	the flow run ID	required

Source code in prefect/input/actions.py

@sync_compatible
@inject_client
async def read_flow_run_input(
    key: str, flow_run_id: Optional[UUID] = None, client: "PrefectClient" = None
) -> Any:
    """Read a flow run input.

    Args:
        - key (str): the flow run input key
        - flow_run_id (UUID): the flow run ID
    """
    flow_run_id = _ensure_flow_run_id(flow_run_id)

    try:
        value = await client.read_flow_run_input(flow_run_id=flow_run_id, key=key)
    except PrefectHTTPStatusError as exc:
        if exc.response.status_code == 404:
            return None
        raise
    else:
        return orjson.loads(value)