fix deserializing callbacks to proper types

[mobuchowski]

Without this fix, the callback requests are deserialized as a base class CallbackRequest and are not executed - they are silently failing through this check.

def _execute_callbacks(
    dagbag: DagBag, callback_requests: list[CallbackRequest], log: FilteringBoundLogger
) -> None:
    for request in callback_requests:
        log.debug("Processing Callback Request", request=request.to_json())
        if isinstance(request, TaskCallbackRequest):
            raise NotImplementedError(
                "Haven't coded Task callback yet - https://github.com/apache/airflow/issues/44354!"
            )
            # _execute_task_callbacks(dagbag, request)
        elif isinstance(request, DagCallbackRequest):
            _execute_dag_callbacks(dagbag, request, log)

[amoghrajesh]

Callbacks haven't yet been ported to TASK SDK @mobuchowski
Is this an attempt to do that?

[mobuchowski]

@amoghrajesh dag ones (DagCallbackRequest) should work, and are mostly implemented in some form. This PR makes basic ones work. There are missing things - like the context is empty dict - but neverthless is improving the situation.

My main motivation is unblocking ticket #44547 that has been really hard to get to get merged.

[kaxil]

few nits

[jscheffl]

I assume this PR broke main, failed tests are with https://github.com/apache/airflow/actions/runs/12657999447

Someone able to fix this or shall we revert to fix canary again?

[kaxil]

I assume this PR broke main, failed tests are with https://github.com/apache/airflow/actions/runs/12657999447

Someone able to fix this or shall we revert to fix canary again?

Fixed in #45482