[AIRFLOW-3801] Fix DagBag collect dags invocation to prevent examples to be loaded (#4677)

airflow/models/__init__.py

@@ -304,7 +304,7 @@ class DagBag(BaseDagBag, LoggingMixin):
         self.import_errors = {}
         self.has_logged = False
 
-        self.collect_dags(dag_folder, include_examples)
+        self.collect_dags(dag_folder=dag_folder, include_examples=include_examples)
 
     def size(self):
         """
@@ -559,7 +559,7 @@ class DagBag(BaseDagBag, LoggingMixin):
         stats = []
         FileLoadStat = namedtuple(
             'FileLoadStat', "file duration dag_num task_num dags")
-        for filepath in list_py_file_paths(dag_folder, include_examples):
+        for filepath in list_py_file_paths(dag_folder, include_examples=include_examples):
             try:
                 ts = timezone.utcnow()
                 found_dags = self.process_file(

tests/models.py

@@ -1471,6 +1471,14 @@ class DagBagTest(unittest.TestCase):
         non_existing_dag_id = "non_existing_dag_id"
         self.assertIsNone(dagbag.get_dag(non_existing_dag_id))
 
+    def test_dont_load_example(self):
+        """
+        test that the example are not loaded
+        """
+        dagbag = models.DagBag(dag_folder=self.empty_dir, include_examples=False)
+
+        self.assertEqual(dagbag.size(), 0)
+
     def test_process_file_that_contains_multi_bytes_char(self):
         """
         test that we're able to parse file that contains multi-byte char