Deserialize should work while retrieving variables with secrets backend

[amoghrajesh]

closes: #50666

Problem

As reported in #50666, deserialize_json option wasn't working when trying to use it with Variable.get with AWS secrets manager.

But this problem on further investigation revealed to be not just for AWS secrets backend but a bug in the Variable module itself. The bug didn't honor the deserialize_json flag at all when it was passed for secrets backend which shouldn't be the case at all.

To prove this, did the following:

Set up a airflow.secrets.local_filesystem.LocalFilesystemBackend and checked with both Airflow 2 and Airflow 3.

var.json

{
    "A": "{\"num1\": 23, \"num2\": 42}"
}

Dag used:

from __future__ import annotations

from airflow.models import Variable
from airflow.models.baseoperator import BaseOperator
from airflow import DAG
# from airflow.sdk import Variable

class CustomOperator(BaseOperator):
    def execute(self, context):
        v = Variable.get(key="A")
        print("Got value", v, type(v))


        v1 = Variable.get(key="A", deserialize_json=True)
        print("Got value from deser", v1, type(v1))


with DAG("get-var-sdk", schedule=None, catchup=False) as dag:
    CustomOperator(task_id="set_var")

Airflow 2:

# With secrets backned
# [2025-05-20, 16:38:47 IST] {logging_mixin.py:190} INFO - Got value {"num1": 23, "num2": 42} <class 'str'>
# [2025-05-20, 16:38:47 IST] {logging_mixin.py:190} INFO - Got value from deser {'num1': 23, 'num2': 42} <class 'dict'>

Airflow 3:

[2025-05-21, 15:35:41] INFO - Got value {"num1": 23, "num2": 42} <class 'str'>: chan="stdout": source="task"
[2025-05-21, 15:35:41] INFO - Got value from deser {"num1": 23, "num2": 42} <class 'str'>: chan="stdout": source="task"

As observed the second print for Airflow 3 is a string when it should really be a dictionary.

Testing

Set up a airflow.secrets.local_filesystem.LocalFilesystemBackend and checked with both Airflow 2 and Airflow 3.

var.json

{
    "A": "{\"num1\": 23, \"num2\": 42}"
}

And exporting these variables in init.sh of breeze:

export AIRFLOW__SECRETS__BACKEND="airflow.secrets.local_filesystem.LocalFilesystemBackend"
export AIRFLOW__SECRETS__BACKEND_KWARGS='{"connections_file_path": "/files/conn.json", "variables_file_path": "/files/var.json"}'

Dag used:

from __future__ import annotations

from airflow.models import Variable
from airflow.models.baseoperator import BaseOperator
from airflow import DAG
# from airflow.sdk import Variable

class CustomOperator(BaseOperator):
    def execute(self, context):
        v = Variable.get(key="A")
        print("Got value", v, type(v))


        v1 = Variable.get(key="A", deserialize_json=True)
        print("Got value from deser", v1, type(v1))


with DAG("get-var-sdk", schedule=None, catchup=False) as dag:
    CustomOperator(task_id="set_var")

Before changes:

After changes:

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[ashb]

Does {{ var.json.myvar }} still work as it did before?

[amoghrajesh]

@ashb yeah it works like it used to.

Tried with:

Variable.set("key", "{\"num1\": 23, \"num2\": 42}")

get_var_task = BashOperator(
    task_id="get_variable_with_template",
    bash_command="echo {{ var.value.key }} && echo {{ var.json.key }}",
    dag=dag
)

Output:

[2025-05-21, 18:28:46] INFO - Running command: ['/usr/bin/bash', '-c', 'echo {"num1": 23, "num2": 42} && echo {\'num1\': 23, \'num2\': 42}']: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-05-21, 18:28:46] INFO - Output:: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-05-21, 1[8](http://localhost:28080/dags/set_get_airflow_variable/runs/manual__2025-05-21T12:58:45.415124+00:00/tasks/get_variable_with_template?try_number=1#8):28:46] INFO - {num1: 23, num2: 42}: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-05-21, 18:28:46] INFO - {num1: 23, num2: 42}: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"

[amoghrajesh]

Unrelated failures, merging.

[github-actions]

Backport successfully created: v3-0-test

Status	Branch	Result
✅	v3-0-test