[AIRFLOW-2778] Bad Import in collect_dag in DagBag

• Type: Bug
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 2.0.0
• Fix Version/s: 1.10.1, 2.0.0
• Component/s: DAG
• Assignee: Kevin Yuen
• Reporter: Kevin Yuen
• Created: 21/Jul/18 16:30
• Updated: 03/Sep/18 13:18
• Resolved: 22/Jul/18 12:17

Description

I run the following test to make sure there are no import errors in our CI. After upgrading to 1.10-stable from 1.9.0, the following test starting failing: 

class TestAirflowDag:

    def test_dagbag_import(self):
        """Verify that Airflow will be able to import all DAGS in the repo
        """

        dagbag = self._get_dagbag()
        assert len(dagbag.import_errors) == 0

    def _get_dagbag(self):
        dag_folder = os.getenv('AIRFLOW_DAGS', False)
        assert dag_folder is not None

        return DagBag(dag_folder=dag_folder, include_examples=False)

The following error was raised: 

# Used to store stats around DagBag processing 
stats = [] 
FileLoadStat = namedtuple( 
    'FileLoadStat', 
    "file duration dag_num task_num dags"
)  
for filepath in utils.dag_processing.list_py_file_paths(dag_folder): 

E AttributeError: module 'airflow.utils' has no attribute 'dag_processing'

The issue is likely because `dag_processing` was not imported. 
https://github.com/apache/incubator-airflow/blob/7f2bc0ddf74c9bf9113401c4ecb1355e6c2fab7f/airflow/models.py#L68-L94

This was further validated by stopping execution using pdb, and running before utils.dag_processing.list_py_file_paths(dag_folder) is called.

>> from airflow.utils.dag_processing import *

The test then proceed to pass since the file was loaded. 

 

Issue Links

• links to: [Github] Pull Request #3624 (KevYuen)