from airflow.sdk import Variable, dag, task

@dag
def ample_simplest_dag():

    @task
    def my_task():
        print("hellooo")
        import os
        os.environ["AIRFLOW_VAR_MY_VARIABLE"] = "my_value"
        assert Variable.get("my_variable") == "my_value"

    my_task()


d = ample_simplest_dag()

root@75043446567f:/opt/airflow# airflow tasks test ample_simplest_dag my_task
[2025-05-20T18:00:14.130+0000] {manager.py:147} INFO - DAG bundles loaded: dags-folder, example_dags, example_standard_dags
[2025-05-20T18:00:14.996+0000] {dagbag.py:573} INFO - Filling up the DagBag from /files/dags
[2025-05-20T18:00:16.315+0000] {dag.py:2321} INFO - created dagrun <DagRun ample_simplest_dag @ None: __airflow_temporary_run_2025-05-20T18:00:16.238529+00:00__, state:running, queued_at: None. run_type: manual>
[2025-05-20T18:00:16.336+0000] {dag.py:1221} INFO - [DAG TEST] starting task_id=my_task map_index=-1
[2025-05-20T18:00:16.336+0000] {dag.py:1224} INFO - [DAG TEST] running task <TaskInstance: ample_simplest_dag.my_task __airflow_temporary_run_2025-05-20T18:00:16.238529+00:00__ [None]>
2025-05-20 18:00:17 [debug    ] Starting task instance run     hostname=75043446567f pid=147 ti_id=0196eeda-74bb-7bd0-8013-23f704571fd4 unixname=root
2025-05-20 18:00:17 [debug    ] Retrieved task instance details dag_id=ample_simplest_dag state=queued task_id=my_task ti_id=0196eeda-74bb-7bd0-8013-23f704571fd4
2025-05-20 18:00:17 [info     ] Task started                   hostname=75043446567f previous_state=queued ti_id=0196eeda-74bb-7bd0-8013-23f704571fd4
2025-05-20 18:00:17 [info     ] Task instance state updated    rows_affected=1 ti_id=0196eeda-74bb-7bd0-8013-23f704571fd4
[2025-05-20T18:00:17.055+0000] {_client.py:1026} INFO - HTTP Request: PATCH http://in-process.invalid./task-instances/0196eeda-74bb-7bd0-8013-23f704571fd4/run "HTTP/1.1 200 OK"
2025-05-20 18:00:17 [debug    ] Sending request                [task] msg=SetRenderedFields(rendered_fields={'templates_dict': None, 'op_args': [], 'op_kwargs': {}}, type='SetRenderedFields')
2025-05-20 18:00:17 [debug    ] Received message from task runner [task] msg=SetRenderedFields(rendered_fields={'templates_dict': None, 'op_args': [], 'op_kwargs': {}}, type='SetRenderedFields')
2025-05-20 18:00:17 [info     ] Updating RenderedTaskInstanceFields field_count=3 ti_id=0196eeda-74bb-7bd0-8013-23f704571fd4
2025-05-20 18:00:17 [debug    ] RenderedTaskInstanceFields updated successfully ti_id=0196eeda-74bb-7bd0-8013-23f704571fd4
[2025-05-20T18:00:17.082+0000] {_client.py:1026} INFO - HTTP Request: PUT http://in-process.invalid./task-instances/0196eeda-74bb-7bd0-8013-23f704571fd4/rtif "HTTP/1.1 201 Created"
Task instance is in running state
 Previous state of the Task instance: TaskInstanceState.QUEUED
Current task name:my_task
Dag name:ample_simplest_dag
hellooo
[2025-05-20T18:00:17.086+0000] {python.py:219} INFO - Done. Returned value was: None
2025-05-20 18:00:17 [debug    ] Sending request                [task] msg=SucceedTask(state='success', end_date=datetime.datetime(2025, 5, 20, 18, 0, 17, 86745, tzinfo=datetime.timezone.utc), task_outlets=[], outlet_events=[], rendered_map_index=None, type='SucceedTask')
2025-05-20 18:00:17 [debug    ] Received message from task runner [task] msg=SucceedTask(state='success', end_date=datetime.datetime(2025, 5, 20, 18, 0, 17, 86745, tzinfo=datetime.timezone.utc), task_outlets=[], outlet_events=[], rendered_map_index=None, type='SucceedTask')
2025-05-20 18:00:17 [debug    ] Updating task instance state   new_state=success ti_id=0196eeda-74bb-7bd0-8013-23f704571fd4
2025-05-20 18:00:17 [debug    ] Retrieved current task instance state max_tries=0 previous_state=running ti_id=0196eeda-74bb-7bd0-8013-23f704571fd4 try_number=0
2025-05-20 18:00:17 [info     ] Task instance state updated    new_state=success rows_affected=1 ti_id=0196eeda-74bb-7bd0-8013-23f704571fd4
[2025-05-20T18:00:17.106+0000] {_client.py:1026} INFO - HTTP Request: PATCH http://in-process.invalid./task-instances/0196eeda-74bb-7bd0-8013-23f704571fd4/state "HTTP/1.1 204 No Content"
2025-05-20 18:00:17 [debug    ] Running finalizers             [task] ti=RuntimeTaskInstance(id=UUID('0196eeda-74bb-7bd0-8013-23f704571fd4'), task_id='my_task', dag_id='ample_simplest_dag', run_id='__airflow_temporary_run_2025-05-20T18:00:16.238529+00:00__', try_number=0, map_index=-1, hostname='75043446567f', context_carrier=None, task=<Task(_PythonDecoratedOperator): my_task>, max_tries=0, start_date=datetime.datetime(2025, 5, 20, 18, 0, 16, 632534, tzinfo=datetime.timezone.utc), end_date=None, state=<TaskInstanceState.SUCCESS: 'success'>, is_mapped=False, rendered_map_index=None, log_url=None)
Task instance in success state
 Previous state of the Task instance: TaskInstanceState.RUNNING
Task operator:<Task(_PythonDecoratedOperator): my_task>
[2025-05-20T18:00:17.118+0000] {dag.py:1270} INFO - [DAG TEST] end task task_id=my_task map_index=-1