Repository: apache/airflow
File: airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py (Python)

Code before:
class TIDeferredStatePayload(StrictBaseModel):
    """Schema for updating TaskInstance to a deferred state."""

    state: Annotated[
        Literal[IntermediateTIState.DEFERRED],
        # Specify a default in the schema, but not in code, so Pydantic marks it as required.
        WithJsonSchema(
            {
                "type": "string",
                "enum": [IntermediateTIState.DEFERRED],
                "default": IntermediateTIState.DEFERRED,
            }
        ),
    ]
    classpath: str
    trigger_kwargs: Annotated[JsonValue | str, Field(default_factory=dict)]
    """
    Kwargs to pass to the trigger constructor, either a plain dict or an encrypted string.

    Both forms will be passed along to the trigger, the server will not handle either.
    """

    trigger_timeout: timedelta | None = None
    next_method: str
    """The name of the method on the operator to call in the worker after the trigger has fired."""
    next_kwargs: Annotated[JsonValue, Field(default_factory=dict)]
    """
    Kwargs to pass to the above method, either a plain dict or an encrypted string.

    Both forms will be passed along to the TaskSDK upon resume, the server will not handle either.
    """
    rendered_map_index: str | None = None


class TIRescheduleStatePayload(StrictBaseModel):
    """Schema for updating TaskInstance to a up_for_reschedule state."""

    state: Annotated[
        Literal[IntermediateTIState.UP_FOR_RESCHEDULE],
        # Specify a default in the schema, but not in code, so Pydantic marks it as required.
        WithJsonSchema(
            {
                "type": "string",
                "enum": [IntermediateTIState.UP_FOR_RESCHEDULE],
                "default": IntermediateTIState.UP_FOR_RESCHEDULE,
            }
        ),
    ]
    reschedule_date: UtcDateTime
    end_date: UtcDateTime


Review comment (suggestion):
I gave this some thought and why I think no cadwyn migration is needed here is because the change from `dict[str, Any]` to `JsonValue` is essentially widening the acceptable types, not restricting them and hence old clients sending `dict[str, Any]` will still validate because dicts are valid `JsonValue` and also sending a valid `JsonValue` to old client with `dict[str, Any]` would still validate fine.


Code after:
class TIDeferredStatePayload(StrictBaseModel):
    """Schema for updating TaskInstance to a deferred state."""

    state: Annotated[
        Literal[IntermediateTIState.DEFERRED],
        # Specify a default in the schema, but not in code, so Pydantic marks it as required.
        WithJsonSchema(
            {
                "type": "string",
                "enum": [IntermediateTIState.DEFERRED],
                "default": IntermediateTIState.DEFERRED,
            }
        ),
    ]
    classpath: str
    trigger_kwargs: Annotated[dict[str, JsonValue] | str, Field(default_factory=dict)]
    """
    Kwargs to pass to the trigger constructor, either a plain dict or an encrypted string.

    Both forms will be passed along to the trigger, the server will not handle either.
    """

    trigger_timeout: timedelta | None = None
    next_method: str
    """The name of the method on the operator to call in the worker after the trigger has fired."""
    next_kwargs: Annotated[dict[str, JsonValue], Field(default_factory=dict)]
    """
    Kwargs to pass to the above method, either a plain dict or an encrypted string.

    Both forms will be passed along to the TaskSDK upon resume, the server will not handle either.
    """
    rendered_map_index: str | None = None


class TIRescheduleStatePayload(StrictBaseModel):
    """Schema for updating TaskInstance to a up_for_reschedule state."""

    state: Annotated[
        Literal[IntermediateTIState.UP_FOR_RESCHEDULE],
        # Specify a default in the schema, but not in code, so Pydantic marks it as required.
        WithJsonSchema(
            {
                "type": "string",
                "enum": [IntermediateTIState.UP_FOR_RESCHEDULE],
                "default": IntermediateTIState.UP_FOR_RESCHEDULE,
            }
        ),
    ]
    reschedule_date: UtcDateTime
    end_date: UtcDateTime