[airflow] 28/32: Add getimport for xcom change

[ka...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

kaxilnaik pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit 1a8ba6a95696a19eb422a3d0a9553b3a59827052
Author: Daniel Imberman <da...@gmail.com>
AuthorDate: Mon Aug 3 15:59:40 2020 -0700

    Add getimport for xcom change
---
 airflow/configuration.py | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/airflow/configuration.py b/airflow/configuration.py
index d912898..01ee90f 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -42,6 +42,7 @@ import yaml
 from zope.deprecation import deprecated
 
 from airflow.exceptions import AirflowConfigException
+from airflow.utils.module_loading import import_string
 
 standard_library.install_aliases()
 
@@ -342,6 +343,26 @@ class AirflowConfigParser(ConfigParser):
                 "section/key [{section}/{key}] not found "
                 "in config".format(section=section, key=key))
 
+    def getimport(self, section, key, **kwargs):
+        """
+        Reads options, imports the full qualified name, and returns the object.
+        In case of failure, it throws an exception a clear message with the key aad the section names
+        :return: The object or None, if the option is empty
+        """
+        full_qualified_path = conf.get(section=section, key=key, **kwargs)
+        if not full_qualified_path:
+            return None
+
+        try:
+            return import_string(full_qualified_path)
+        except ImportError as e:
+            log.error(e)
+            raise AirflowConfigException(
+                'The object could not be loaded. Please check "{key}" key in "{section}" section. '
+                'Current value: "{full_qualified_path}".' .format(
+                    key=key, section=section, full_qualified_path=full_qualified_path)
+            )
+
     def getboolean(self, section, key, **kwargs):
         val = str(self.get(section, key, **kwargs)).lower().strip()
         if '#' in val: