DateTimeSensorAsync is failing

[atul-astronomer]

Apache Airflow version

main (development)

If "Other Airflow 2 version" selected, which one?

No response

What happened?

ERROR - Trigger failed:
Traceback (most recent call last):

  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 965, in create_triggers
    trigger_instance = trigger_class(**deserialised_kwargs)

  File "/opt/airflow/providers/standard/src/airflow/providers/standard/triggers/temporal.py", line 51, in __init__
    raise TypeError(f"Expected datetime.datetime type for moment. Got {type(moment)}")

TypeError: Expected datetime.datetime type for moment. Got <class 'str'>
[2025-08-26, 12:44:51] ERROR - Task failed with exception
TaskDeferralError: Trigger failure
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 919 in run

File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1301 in _execute_task

File "/opt/airflow/task-sdk/src/airflow/sdk/bases/sensor.py", line 255 in resume_execution

File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 1614 in resume_execution

What you think should happen instead?

No response

How to reproduce

Run the below dag in main branch:

from airflow.sdk import DAG
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.providers.standard.sensors.date_time import DateTimeSensorAsync
from airflow.utils.trigger_rule import TriggerRule
from pendulum import datetime

with DAG(
    dag_id='example_deferrable_dag',
    start_date=datetime(2023, 1, 1),
    catchup=False,
    tags=['async'],
) as dag:

    start = EmptyOperator(
        task_id='start'
    )

    wait = DateTimeSensorAsync(
        task_id='wait_for_time',
        target_time="""{{ macros.datetime.utcnow() + macros.timedelta(minutes=1) }}"""
    )

    end = EmptyOperator(
        task_id='end'
    )

    start >> wait >> end

Operating System

Linux

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[mandeepzemo]

@atul-astronomer I will check this, Please assign

[amoghrajesh]

CC: @uranusjr seems related to the recent change: #53696?

[vatsrahul1001]

When I was trying this in main I got below trace

[2025-08-26T10:30:04.036+0000] {triggerer_job_runner.py:183} INFO - Exited trigger loop
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 10, in <module>
    sys.exit(main())
  File "/opt/airflow/airflow-core/src/airflow/__main__.py", line 55, in main
    args.func(args)
  File "/opt/airflow/airflow-core/src/airflow/cli/cli_config.py", line 49, in command
    return func(*args, **kwargs)
  File "/opt/airflow/airflow-core/src/airflow/utils/cli.py", line 114, in wrapper
    return f(*args, **kwargs)
  File "/opt/airflow/airflow-core/src/airflow/utils/providers_configuration_loader.py", line 54, in wrapped_function
    return func(*args, **kwargs)
  File "/opt/airflow/airflow-core/src/airflow/cli/commands/triggerer_command.py", line 69, in triggerer
    run_command_with_daemon_option(
  File "/opt/airflow/airflow-core/src/airflow/cli/commands/daemon_utils.py", line 86, in run_command_with_daemon_option
    callback()
  File "/opt/airflow/airflow-core/src/airflow/cli/commands/triggerer_command.py", line 72, in <lambda>
    callback=lambda: triggerer_run(args.skip_serve_logs, args.capacity, triggerer_heartrate),
  File "/opt/airflow/airflow-core/src/airflow/cli/commands/triggerer_command.py", line 55, in triggerer_run
    run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute)
  File "/opt/airflow/airflow-core/src/airflow/utils/session.py", line 100, in wrapper
    return func(*args, session=session, **kwargs)
  File "/opt/airflow/airflow-core/src/airflow/jobs/job.py", line 368, in run_job
    return execute_job(job, execute_callable=execute_callable)
  File "/opt/airflow/airflow-core/src/airflow/jobs/job.py", line 397, in execute_job
    ret = execute_callable()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 172, in _execute
    self.trigger_runner.run()
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 528, in run
    self.load_triggers()
  File "/opt/airflow/airflow-core/src/airflow/traces/tracer.py", line 58, in wrapper
    return func(*args, **kwargs)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 551, in load_triggers
    self.update_triggers(set(ids))
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 685, in update_triggers
    workload = create_workload(new_trigger_orm)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 626, in create_workload
    trigger = expand_start_trigger_args(trigger)
  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 612, in expand_start_trigger_args
    task = dag_bag.get_dag(trigger.task_instance.dag_id).get_task(trigger.task_instance.task_id)
AttributeError: 'NoneType' object has no attribute 'get_task'

[mandeepzemo]

Hi Team,
I made the following changes in the temporal.py to handle string inputs for moment and ensure proper datetime validation:

def __init__(self, moment: datetime.datetime, *, end_from_trigger: bool = False) -> None:
    super().__init__()
    if isinstance(moment, str):
        try:
            moment = datetime.datetime.fromisoformat(moment)
        except (ValueError, TypeError) as e:
            raise ValueError(f"Invalid datetime string format. Expected ISO format. Error: {e}")
    elif not isinstance(moment, datetime.datetime):
        raise TypeError(f"Expected datetime.datetime type for moment. Got {type(moment)}")
    # Make sure it's in UTC
    if moment.tzinfo is None:
        raise ValueError("You cannot pass naive datetimes")
    self.moment: pendulum.DateTime = timezone.convert_to_utc(moment)
    self.end_from_trigger = end_from_trigger

This fix resolved the error:

Expected datetime.datetime type for moment. Got <class 'str'>

However, after this change, the task is still getting deferred (screenshot attached below).
Could you please confirm if this is the expected behavior, or if additional changes are required to prevent the deferral?

Thanks in advance for your guidance.

[vatsrahul1001]

@atul-astronomer can you confirm if this is still happening in main

[atul-astronomer]

Reverified on main, issue is fixed now. @vatsrahul1001