BasPH commented on a change in pull request #4880: [AIRFLOW-3735] Separate dag
parsing from checking
URL: https://github.com/apache/airflow/pull/4880#discussion_r263812324
##########
File path: airflow/models/__init__.py
##########
@@ -369,105 +369,36 @@ def process_file(self, filepath, only_if_updated=True,
safe_mode=True):
return found_dags
except Exception as e:
+ self.import_errors[filepath] = str(e)
self.log.exception(e)
return found_dags
- mods = []
- is_zipfile = zipfile.is_zipfile(filepath)
- if not is_zipfile:
- if safe_mode and os.path.isfile(filepath):
- with open(filepath, 'rb') as f:
- content = f.read()
- if not all([s in content for s in (b'DAG', b'airflow')]):
- self.file_last_changed[filepath] =
file_last_changed_on_disk
- # Don't want to spam user with skip messages
- if not self.has_logged:
- self.has_logged = True
- self.log.info(
- "File %s assumed to contain no DAGs.
Skipping.",
- filepath)
- return found_dags
-
- self.log.debug("Importing %s", filepath)
- org_mod_name, _ = os.path.splitext(os.path.split(filepath)[-1])
- mod_name = ('unusual_prefix_' +
- hashlib.sha1(filepath.encode('utf-8')).hexdigest() +
- '_' + org_mod_name)
-
- if mod_name in sys.modules:
- del sys.modules[mod_name]
-
- with timeout(configuration.conf.getint('core',
"DAGBAG_IMPORT_TIMEOUT")):
- try:
- m = imp.load_source(mod_name, filepath)
- mods.append(m)
- except Exception as e:
- self.log.exception("Failed to import: %s", filepath)
- self.import_errors[filepath] = str(e)
- self.file_last_changed[filepath] =
file_last_changed_on_disk
-
- else:
- zip_file = zipfile.ZipFile(filepath)
- for mod in zip_file.infolist():
- head, _ = os.path.split(mod.filename)
- mod_name, ext = os.path.splitext(mod.filename)
- if not head and (ext == '.py' or ext == '.pyc'):
- if mod_name == '__init__':
- self.log.warning("Found __init__.%s at root of %s",
ext, filepath)
- if safe_mode:
- with zip_file.open(mod.filename) as zf:
- self.log.debug("Reading %s from %s", mod.filename,
filepath)
- content = zf.read()
- if not all([s in content for s in (b'DAG',
b'airflow')]):
- self.file_last_changed[filepath] = (
- file_last_changed_on_disk)
- # todo: create ignore list
- # Don't want to spam user with skip messages
- if not self.has_logged:
- self.has_logged = True
- self.log.info(
- "File %s assumed to contain no DAGs.
Skipping.",
- filepath)
-
- if mod_name in sys.modules:
- del sys.modules[mod_name]
+ try:
+ found_dags = process_dag_file(filepath, safe_mode=safe_mode)
+ except Exception as e:
+ self.import_errors[filepath] = str(e)
+ self.log.exception("Failed to dag file: %s, message: $s",
filepath, e)
Review comment:
I think you mean "Failed to process DAG file:"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
