Cast macro datetime string inputs explicitly

[uranusjr]

Finishing and close #19249. Fix #19241.

This coerce template variables to strings to work around a limitation in lazy_object_proxy when used against built-in types.

Some tests against macro functions are added to ensure macro functions all work against lazy object proxy objects and avoid regression in the future.

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[potiuk]

I think test_airflow_context needs fixing @uranusjr

[uranusjr]

Weird, this PR does not change anything related to that. I’ll rebase to latest and see if that fixes things...

[potiuk]

I'd feel safer if it is rebased again after the fix from Ash :)

[abhishekshenoy]

@potiuk We recently upgraded to 2.2.2 and are facing this issue in our dags which have some macros as below :
SOURCE_OBJECT = "{{ var.json.bucket_path }}/recvd_dt={{ macros.ds_add(next_ds, -1) }}/"

Is upgrading to 2.2.3 the only solution ?

[potiuk]

Is upgrading to 2.2.3 the only solution ?

It's certainly the best.

[abhishekshenoy]

@potiuk Upgrading will be an issue as it involves a long wait time. Have come up with the below 2 approaches as a fix for the time being.

• Using Xcom

    def getRecvdDate(**kwargs):
        next_exec_date = kwargs.get('templates_dict').get('next_exec_date')
        days_delta = kwargs.get('templates_dict').get('time_delta')
        dt=datetime.strptime(next_exec_date, '%Y-%m-%d') + timedelta(days=float(days_delta))
        kwargs['ti'].xcom_push(key='recvd_date', value=dt.strftime("%Y-%m-%d"))
        return


    generate_next_exec_date_minus_one = PythonOperator(task_id='generate_next_exec_date_minus_one',
                        provide_context=True,
                        python_callable=getRecvdDate,
                        templates_dict={'next_exec_date': '{{ next_ds }}',
                                        'time_delta': '{{ var.json.macros_test.time_delta }}'},
                        dag=dag)


    xcom_based_recvd_date = GCSUploadSessionCompleteSensor(
        task_id='xcom_based_recvd_date',
        bucket=BUCKET_NAME,
        prefix="macros-test/recvd_dt={{ task_instance.xcom_pull(task_ids='generate_next_exec_date_minus_one', key='recvd_date') }}/",
        impersonation_chain=IMPERSONATE_SERVICE_ACCOUNT,
        timeout=1,
        soft_fail=True,
        dag=dag
    )

• Using user-defined-macro

import croniter
 
sched = '0 6 * * sat'    

Define below under dag_args :
'user_defined_macros': {
    'custom_next_exec_date': next_exec_date,
}
 
def next_exec_date(dt,days_delta):
    cron = croniter.croniter(sched, dt)
    return cron.get_next(datetime) + timedelta(days=days_delta) 
 
SOURCE_OBJECT = "{{ var.json.macros-test.sourcedir }}/recvd_dt={{ ( custom_next_exec_date(ds,-1) ) }}/"
 
# Task 1: Checking the directory for file availability
file_check = GCSObjectsWtihPrefixExistenceSensor(
    task_id="check_current_date_directory_existence",
    bucket=INBOX_BUCKET,
    prefix=SOURCE_OBJECT,
    impersonation_chain=GCS_IMPERSONATE_SERVICE_ACCOUNT,
    timeout=5,
    soft_fail=False,
    dag=dag
)

[potiuk]

@potiuk Upgrading will be an issue as it involves a long wait time. Have come up with the below 2 approaches as a fix for the time being.

Sure you can come up with your own solutions if "not upgrading to a released patchlevel that contains a fix" is your own choice. But please don't ask maintainers to come up with workarounds if they already provided a fix that you can easily install.

There is no way maintainers can give you another answer here - asking for it is quite unreasonable, because - the fix is there.

What happens if (say) tomorrow we release 2.2.4 with a critical security fix (like Log4j) ? Would you take the risk of scrambling and trying to upgrade everything under the pressure of not only your security team but also your boss who will insist on fixing it "now".

By not upgrading to latest released patchlevel, especially when it contains a fix to a known problem you have and opting to choose a workaround instead, you make yourself vulnerable to this scenario. But do not ask maintainers to either help you with it or endorse it, because we would be shooting ourselves in a foot (and your foot) by suggesting it.

Of course, you are free to do what you want - we cannot stop you. But you've been warned.

BTW. No 2.2.3 does not have any long migrations comparing to 2.2.2: https://airflow.apache.org/docs/apache-airflow/stable/migrations-ref.html?highlight=migrations it just adds one column to one table which should be almost instantenous. I'd really, really recommend you to upgrade.

[abhishekshenoy]

I understand that airflow version upgrade is the optimal solution for this and we will be doing the same in dev . The long wait time is not w.r.t to Airflow migration in itself but certain process that we need to adhere to , any version upgrade needs to go under sanity test in dev for a period of time before it is pushed to prod.

[potiuk]

I can only sympathise with your pain.

But it does not change my answer.

[potiuk]

Suggestion (this is the advice I have from one of the best Project Managers I worked with)

Upgrade often, upgrade as early as you get new version released. This way your processes to upgrade get perfected, and you will figure out how to do them faster. And when the time comes you need to upgrade in 2 hours when there is an urgent security fix - you will.