Use of `conn.extra_dejson` in Airflow 3.0.5 fails with NotImplementedError error

[ashb]

after upgrading to airflow 3.0.5 operators/hooks failed with similar messages:

[2025-08-21, 07:05:11] INFO - Connection Retrieved 'snowflake_default': source="airflow.hooks.base"
[2025-08-21, 07:05:11] ERROR - Task failed with exception: source="task"
NotImplementedError: Objects of type <class 'pydantic_core._pydantic_core.SerializationIterator'> are not supported
File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 918 in run

File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1213 in _execute_task

File "/usr/local/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 397 in wrapper

File "/usr/local/lib/python3.12/site-packages/airflow/providers/common/sql/operators/sql.py", line 307 in execute

File "/usr/local/lib/python3.12/site-packages/airflow/providers/common/sql/operators/sql.py", line 201 in get_db_hook

File "/usr/local/lib/python3.12/functools.py", line 998 in __get__

File "/usr/local/lib/python3.12/site-packages/airflow/providers/common/sql/operators/sql.py", line 177 in _hook

File "/usr/local/lib/python3.12/site-packages/airflow/providers/common/sql/operators/sql.py", line 166 in get_hook

File "/usr/local/lib/python3.12/site-packages/airflow/sdk/definitions/connection.py", line 162 in extra_dejson

File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/secrets_masker.py", line 130 in mask_secret

File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/comms.py", line 187 in send

File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/comms.py", line 145 in as_bytes

File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/comms.py", line 122 in _msgpack_enc_hook

Originally posted by @dshvedchenko in #54574 (comment)

[dshvedchenko]

thanks for posting it here @ashb
here is pytest code

def test_getconn():
    from airflow.sdk.execution_time.comms import MaskSecret, _RequestFrame
    import attrs

   # https://github.com/apache/airflow/blob/00b3b14f02af3acafc532f410f5c05d77b7efa58/task-sdk/src/airflow/sdk/execution_time/secrets_masker.py#L127
    msg = MaskSecret(value=dict(a="df"), name=None)

    @attrs.define
    class CM:
        """Context manager to simulate sending a message."""
        id_counter: Iterator[int] = attrs.field(factory=itertools.count)

        def send(self, msg):
            frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
            frame.as_bytes()

    CM().send(msg)

[dshvedchenko]

if we process json secret , then , frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump(mode="json")) must be used to be happy

[dshvedchenko]

can mode be inferred ?? like

            body = msg.model_dump(mode="python" if isinstance(msg.value, str) else "json")
            frame = _RequestFrame(id=next(self.id_counter), body=body)

[ashb]

mode="json" will mess up things like datetimes I think (it'll send it as a string, not as a dattime) so that isn't the best fix

[dshvedchenko]

so we need to use json to array and dict only ?

ok i will wait for your tests finished

[ashb]

Dict values could be datetime etc too, so we can't use that "json" mode either (we aren't sending this as JSON either, so that would be a kludge)

[dshvedchenko]

thank you for quick fixing it @ashb !!!

[potiuk]

so we need to use json to array and dict only ?

ok i will wait for your tests finished

@dshvedchenko => the tests are added in the PR #54769 - 3.0.5 is yanked and we are releasing 3.0.6rc1 with a fix today. It would be great if you could run System tests with the rc :)

[dshvedchenko]

@potiuk yes, i will check them. im using astro dev 3.0-9 , so need to hack dockerfile to alter constraints