Back to Repositories

Testing LatestOnlyOperator Implementation in Apache Airflow

This test suite demonstrates the implementation and functionality of the LatestOnlyOperator in Apache Airflow. The example showcases how to ensure tasks only run for the most recent scheduled interval, which is crucial for workflows that should only process current data.

Test Coverage Overview

The test coverage focuses on validating the LatestOnlyOperator’s core functionality in an Airflow DAG.

Key areas covered include:

DAG scheduling with 4-hour intervals
Task dependency management between LatestOnlyOperator and EmptyOperator
Catchup behavior configuration
Integration with task scheduling mechanics

Implementation Analysis

The testing approach utilizes a straightforward DAG structure to demonstrate the LatestOnlyOperator’s behavior.

Technical implementation details:

DAG configuration using context manager syntax
Explicit scheduling interval definition
Task dependency definition using bitshift operator
Tag-based organization for example categorization

Technical Details

Testing environment specifications:

Python’s datetime module for temporal logic
Airflow’s DAG and operator imports
EmptyOperator for downstream task simulation
Standard provider package integration
Catchup disabled for simplified execution

Best Practices Demonstrated

The test implementation showcases several Airflow best practices for operator usage and DAG design.

Notable practices include:

Clear task ID naming conventions
Proper import organization
Explicit schedule interval definition
Appropriate use of DAG tags for categorization
Clean dependency chain definition

apache/airflow

airflow/example_dags/example_latest_only.py

            
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
"""Example of the LatestOnlyOperator"""

from __future__ import annotations

import datetime

from airflow.models.dag import DAG
from airflow.operators.empty import EmptyOperator
from airflow.providers.standard.operators.latest_only import LatestOnlyOperator

with DAG(
    dag_id="latest_only",
    schedule=datetime.timedelta(hours=4),
    start_date=datetime.datetime(2021, 1, 1),
    catchup=False,
    tags=["example2", "example3"],
) as dag:
    latest_only = LatestOnlyOperator(task_id="latest_only")
    task1 = EmptyOperator(task_id="task1")

    latest_only >> task1