You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by as...@apache.org on 2021/04/15 11:28:41 UTC
[airflow] branch v2-0-test updated (3634738 -> ef876cf)
This is an automated email from the ASF dual-hosted git repository.
ash pushed a change to branch v2-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git.
from 3634738 Unable to trigger backfill or manual jobs with Kubernetes executor. (#14160)
new 894c646 Restore base lineage backend (#14146)
new cf3de8f Add documentation create/update community providers (#15061)
new dc9bf44 Fix url generation for TriggerDagRunOperatorLink (#14990)
new cfeeb14 Fix password masking in CLI action_logging (#15143)
new 315005b BugFix: CLI 'kubernetes cleanup-pods' should only clean up Airflow-created Pods (#15204)
new 7e28c97 Change default of `[kubernetes] enable_tcp_keepalive` to `True` (#15338)
new 7b9f091 Fix missing on_load trigger for folder-based plugins (#15208)
new ef876cf Import Connection lazily in hooks to avoid cycles (#15361)
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
airflow/cli/cli_parser.py | 6 +-
airflow/cli/commands/kubernetes_command.py | 18 +-
airflow/config_templates/config.yml | 2 +-
airflow/config_templates/default_airflow.cfg | 2 +-
airflow/hooks/base.py | 12 +-
airflow/kubernetes/kube_client.py | 8 +-
airflow/lineage/__init__.py | 22 ++
airflow/lineage/backend.py | 47 ++++
airflow/plugins_manager.py | 23 +-
airflow/utils/cli.py | 20 +-
airflow/utils/helpers.py | 4 +-
.../howto/create-update-providers.rst | 301 +++++++++++++++++++++
docs/apache-airflow-providers/index.rst | 14 +-
docs/apache-airflow/lineage.rst | 21 ++
tests/cli/commands/test_kubernetes_command.py | 40 ++-
tests/lineage/test_lineage.py | 49 +++-
tests/plugins/test_plugin.py | 7 +
tests/plugins/test_plugins_manager.py | 49 ++++
tests/utils/test_cli_util.py | 10 +
tests/utils/test_helpers.py | 13 +-
20 files changed, 624 insertions(+), 44 deletions(-)
create mode 100644 airflow/lineage/backend.py
create mode 100644 docs/apache-airflow-providers/howto/create-update-providers.rst
[airflow] 05/08: BugFix: CLI 'kubernetes cleanup-pods' should only
clean up Airflow-created Pods (#15204)
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ash pushed a commit to branch v2-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 315005b2ffe7b180f5d4f873278606fd8ffaf1ca
Author: Xiaodong DENG <xd...@apache.org>
AuthorDate: Thu Apr 8 12:11:08 2021 +0200
BugFix: CLI 'kubernetes cleanup-pods' should only clean up Airflow-created Pods (#15204)
closes: #15193
Currently condition if the pod is created by Airflow is not considered. This commit fixes this.
We decide if the Pod is created by Airflow via checking if it has all the labels added in PodGenerator.construct_pod() or KubernetesPodOperator.create_labels_for_pod().
(cherry picked from commit c594d9cfb32bbcfe30af3f5dcb452c6053cacc95)
---
airflow/cli/cli_parser.py | 6 +++-
airflow/cli/commands/kubernetes_command.py | 18 +++++++++++-
tests/cli/commands/test_kubernetes_command.py | 40 +++++++++++++++++++++------
3 files changed, 54 insertions(+), 10 deletions(-)
diff --git a/airflow/cli/cli_parser.py b/airflow/cli/cli_parser.py
index c33a854..b5384b4 100644
--- a/airflow/cli/cli_parser.py
+++ b/airflow/cli/cli_parser.py
@@ -1341,7 +1341,11 @@ CONFIG_COMMANDS = (
KUBERNETES_COMMANDS = (
ActionCommand(
name='cleanup-pods',
- help="Clean up Kubernetes pods in evicted/failed/succeeded states",
+ help=(
+ "Clean up Kubernetes pods "
+ "(created by KubernetesExecutor/KubernetesPodOperator) "
+ "in evicted/failed/succeeded states"
+ ),
func=lazy_load_command('airflow.cli.commands.kubernetes_command.cleanup_pods'),
args=(ARG_NAMESPACE,),
),
diff --git a/airflow/cli/commands/kubernetes_command.py b/airflow/cli/commands/kubernetes_command.py
index f98c45e..daf11a3 100644
--- a/airflow/cli/commands/kubernetes_command.py
+++ b/airflow/cli/commands/kubernetes_command.py
@@ -90,7 +90,23 @@ def cleanup_pods(args):
print('Loading Kubernetes configuration')
kube_client = get_kube_client()
print(f'Listing pods in namespace {namespace}')
- list_kwargs = {"namespace": namespace, "limit": 500}
+ airflow_pod_labels = [
+ 'dag_id',
+ 'task_id',
+ 'execution_date',
+ 'try_number',
+ 'airflow_version',
+ ]
+ list_kwargs = {
+ "namespace": namespace,
+ "limit": 500,
+ "label_selector": client.V1LabelSelector(
+ match_expressions=[
+ client.V1LabelSelectorRequirement(key=label, operator="Exists")
+ for label in airflow_pod_labels
+ ]
+ ),
+ }
while True: # pylint: disable=too-many-nested-blocks
pod_list = kube_client.list_namespaced_pod(**list_kwargs)
for pod in pod_list.items:
diff --git a/tests/cli/commands/test_kubernetes_command.py b/tests/cli/commands/test_kubernetes_command.py
index 8ae2eef..707eb55 100644
--- a/tests/cli/commands/test_kubernetes_command.py
+++ b/tests/cli/commands/test_kubernetes_command.py
@@ -55,6 +55,13 @@ class TestGenerateDagYamlCommand(unittest.TestCase):
class TestCleanUpPodsCommand(unittest.TestCase):
+ label_selector = kubernetes.client.V1LabelSelector(
+ match_expressions=[
+ kubernetes.client.V1LabelSelectorRequirement(key=label, operator="Exists")
+ for label in ['dag_id', 'task_id', 'execution_date', 'try_number', 'airflow_version']
+ ]
+ )
+
@classmethod
def setUpClass(cls):
cls.parser = cli_parser.get_parser()
@@ -79,7 +86,9 @@ class TestCleanUpPodsCommand(unittest.TestCase):
kubernetes_command.cleanup_pods(
self.parser.parse_args(['kubernetes', 'cleanup-pods', '--namespace', 'awesome-namespace'])
)
- list_namespaced_pod.assert_called_once_with(namespace='awesome-namespace', limit=500)
+ list_namespaced_pod.assert_called_once_with(
+ namespace='awesome-namespace', limit=500, label_selector=self.label_selector
+ )
delete_pod.assert_not_called()
load_incluster_config.assert_called_once()
@@ -98,7 +107,9 @@ class TestCleanUpPodsCommand(unittest.TestCase):
kubernetes_command.cleanup_pods(
self.parser.parse_args(['kubernetes', 'cleanup-pods', '--namespace', 'awesome-namespace'])
)
- list_namespaced_pod.assert_called_once_with(namespace='awesome-namespace', limit=500)
+ list_namespaced_pod.assert_called_once_with(
+ namespace='awesome-namespace', limit=500, label_selector=self.label_selector
+ )
delete_pod.assert_called_with('dummy', 'awesome-namespace')
load_incluster_config.assert_called_once()
@@ -120,7 +131,9 @@ class TestCleanUpPodsCommand(unittest.TestCase):
kubernetes_command.cleanup_pods(
self.parser.parse_args(['kubernetes', 'cleanup-pods', '--namespace', 'awesome-namespace'])
)
- list_namespaced_pod.assert_called_once_with(namespace='awesome-namespace', limit=500)
+ list_namespaced_pod.assert_called_once_with(
+ namespace='awesome-namespace', limit=500, label_selector=self.label_selector
+ )
delete_pod.assert_not_called()
load_incluster_config.assert_called_once()
@@ -142,7 +155,9 @@ class TestCleanUpPodsCommand(unittest.TestCase):
kubernetes_command.cleanup_pods(
self.parser.parse_args(['kubernetes', 'cleanup-pods', '--namespace', 'awesome-namespace'])
)
- list_namespaced_pod.assert_called_once_with(namespace='awesome-namespace', limit=500)
+ list_namespaced_pod.assert_called_once_with(
+ namespace='awesome-namespace', limit=500, label_selector=self.label_selector
+ )
delete_pod.assert_called_with('dummy3', 'awesome-namespace')
load_incluster_config.assert_called_once()
@@ -162,7 +177,9 @@ class TestCleanUpPodsCommand(unittest.TestCase):
kubernetes_command.cleanup_pods(
self.parser.parse_args(['kubernetes', 'cleanup-pods', '--namespace', 'awesome-namespace'])
)
- list_namespaced_pod.assert_called_once_with(namespace='awesome-namespace', limit=500)
+ list_namespaced_pod.assert_called_once_with(
+ namespace='awesome-namespace', limit=500, label_selector=self.label_selector
+ )
delete_pod.assert_called_with('dummy4', 'awesome-namespace')
load_incluster_config.assert_called_once()
@@ -182,7 +199,9 @@ class TestCleanUpPodsCommand(unittest.TestCase):
kubernetes_command.cleanup_pods(
self.parser.parse_args(['kubernetes', 'cleanup-pods', '--namespace', 'awesome-namespace'])
)
- list_namespaced_pod.assert_called_once_with(namespace='awesome-namespace', limit=500)
+ list_namespaced_pod.assert_called_once_with(
+ namespace='awesome-namespace', limit=500, label_selector=self.label_selector
+ )
load_incluster_config.assert_called_once()
@mock.patch('airflow.cli.commands.kubernetes_command._delete_pod')
@@ -204,8 +223,13 @@ class TestCleanUpPodsCommand(unittest.TestCase):
self.parser.parse_args(['kubernetes', 'cleanup-pods', '--namespace', 'awesome-namespace'])
)
calls = [
- call.first(namespace='awesome-namespace', limit=500),
- call.second(namespace='awesome-namespace', limit=500, _continue='dummy-token'),
+ call.first(namespace='awesome-namespace', limit=500, label_selector=self.label_selector),
+ call.second(
+ namespace='awesome-namespace',
+ limit=500,
+ label_selector=self.label_selector,
+ _continue='dummy-token',
+ ),
]
list_namespaced_pod.assert_has_calls(calls)
delete_pod.assert_called_with('dummy', 'awesome-namespace')
[airflow] 03/08: Fix url generation for TriggerDagRunOperatorLink
(#14990)
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ash pushed a commit to branch v2-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit dc9bf44a59392af0c59bcee45343ae7073cbd139
Author: Alan Ma <al...@gmail.com>
AuthorDate: Sun Apr 11 04:51:59 2021 -0700
Fix url generation for TriggerDagRunOperatorLink (#14990)
Fixes: #14675
Instead of building the relative url manually, we can leverage flask's url generation to account for differing airflow base URL and HTML base URL.
(cherry picked from commit aaa3bf6b44238241bd61178426b692df53770c22)
---
airflow/utils/helpers.py | 4 +++-
tests/utils/test_helpers.py | 13 ++++++++++---
2 files changed, 13 insertions(+), 4 deletions(-)
diff --git a/airflow/utils/helpers.py b/airflow/utils/helpers.py
index 69ac5a0..7fce177 100644
--- a/airflow/utils/helpers.py
+++ b/airflow/utils/helpers.py
@@ -24,6 +24,7 @@ from itertools import filterfalse, tee
from typing import Any, Callable, Dict, Generator, Iterable, List, Optional, TypeVar
from urllib import parse
+from flask import url_for
from jinja2 import Template
from airflow.configuration import conf
@@ -213,4 +214,5 @@ def build_airflow_url_with_query(query: Dict[str, Any]) -> str:
'http://0.0.0.0:8000/base/graph?dag_id=my-task&root=&execution_date=2020-10-27T10%3A59%3A25.615587
"""
view = conf.get('webserver', 'dag_default_view').lower()
- return f"/{view}?{parse.urlencode(query)}"
+ url = url_for(f"Airflow.{view}")
+ return f"{url}?{parse.urlencode(query)}"
diff --git a/tests/utils/test_helpers.py b/tests/utils/test_helpers.py
index fffa2d4..bb7b453 100644
--- a/tests/utils/test_helpers.py
+++ b/tests/utils/test_helpers.py
@@ -142,10 +142,17 @@ class TestHelpers(unittest.TestCase):
@conf_vars(
{
- ("webserver", "dag_default_view"): "custom",
+ ("webserver", "dag_default_view"): "graph",
}
)
def test_build_airflow_url_with_query(self):
+ """
+ Test query generated with dag_id and params
+ """
query = {"dag_id": "test_dag", "param": "key/to.encode"}
- url = build_airflow_url_with_query(query)
- assert url == "/custom?dag_id=test_dag¶m=key%2Fto.encode"
+ expected_url = "/graph?dag_id=test_dag¶m=key%2Fto.encode"
+
+ from airflow.www.app import cached_app
+
+ with cached_app(testing=True).test_request_context():
+ assert build_airflow_url_with_query(query) == expected_url
[airflow] 01/08: Restore base lineage backend (#14146)
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ash pushed a commit to branch v2-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 894c646386d5aad049796214275fa14411dbcd45
Author: João Ponte <JP...@users.noreply.github.com>
AuthorDate: Sat Apr 3 10:26:59 2021 +0200
Restore base lineage backend (#14146)
This adds back the base lineage backend which can be extended to send lineage metadata to any custom backend.
closes: #14106
Co-authored-by: Joao Ponte <jp...@plista.com>
Co-authored-by: Tomek Urbaszek <tu...@gmail.com>
(cherry picked from commit af2d11e36ed43b0103a54780640493b8ae46d70e)
---
airflow/lineage/__init__.py | 22 ++++++++++++++++++
airflow/lineage/backend.py | 47 +++++++++++++++++++++++++++++++++++++++
docs/apache-airflow/lineage.rst | 21 ++++++++++++++++++
tests/lineage/test_lineage.py | 49 ++++++++++++++++++++++++++++++++++++++++-
4 files changed, 138 insertions(+), 1 deletion(-)
diff --git a/airflow/lineage/__init__.py b/airflow/lineage/__init__.py
index 65f19ef..905eb00 100644
--- a/airflow/lineage/__init__.py
+++ b/airflow/lineage/__init__.py
@@ -25,6 +25,8 @@ import attr
import jinja2
from cattr import structure, unstructure
+from airflow.configuration import conf
+from airflow.lineage.backend import LineageBackend
from airflow.utils.module_loading import import_string
ENV = jinja2.Environment()
@@ -45,6 +47,22 @@ class Metadata:
data: Dict = attr.ib()
+def get_backend() -> Optional[LineageBackend]:
+ """Gets the lineage backend if defined in the configs"""
+ clazz = conf.getimport("lineage", "backend", fallback=None)
+
+ if clazz:
+ if not issubclass(clazz, LineageBackend):
+ raise TypeError(
+ f"Your custom Lineage class `{clazz.__name__}` "
+ f"is not a subclass of `{LineageBackend.__name__}`."
+ )
+ else:
+ return clazz()
+
+ return None
+
+
def _get_instance(meta: Metadata):
"""Instantiate an object from Metadata"""
cls = import_string(meta.type_name)
@@ -82,6 +100,7 @@ def apply_lineage(func: T) -> T:
Saves the lineage to XCom and if configured to do so sends it
to the backend.
"""
+ _backend = get_backend()
@wraps(func)
def wrapper(self, context, *args, **kwargs):
@@ -101,6 +120,9 @@ def apply_lineage(func: T) -> T:
context, key=PIPELINE_INLETS, value=inlets, execution_date=context['ti'].execution_date
)
+ if _backend:
+ _backend.send_lineage(operator=self, inlets=self.inlets, outlets=self.outlets, context=context)
+
return ret_val
return cast(T, wrapper)
diff --git a/airflow/lineage/backend.py b/airflow/lineage/backend.py
new file mode 100644
index 0000000..edfbe0e
--- /dev/null
+++ b/airflow/lineage/backend.py
@@ -0,0 +1,47 @@
+#
+# 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.
+"""Sends lineage metadata to a backend"""
+from typing import TYPE_CHECKING, Optional
+
+if TYPE_CHECKING:
+ from airflow.models.baseoperator import BaseOperator # pylint: disable=cyclic-import
+
+
+class LineageBackend:
+ """Sends lineage metadata to a backend"""
+
+ def send_lineage(
+ self,
+ operator: 'BaseOperator',
+ inlets: Optional[list] = None,
+ outlets: Optional[list] = None,
+ context: Optional[dict] = None,
+ ):
+ """
+ Sends lineage metadata to a backend
+
+ :param operator: the operator executing a transformation on the inlets and outlets
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param inlets: the inlets to this operator
+ :type inlets: list
+ :param outlets: the outlets from this operator
+ :type outlets: list
+ :param context: the current context of the task instance
+ :type context: dict
+ """
+ raise NotImplementedError()
diff --git a/docs/apache-airflow/lineage.rst b/docs/apache-airflow/lineage.rst
index a29f042..362d3e6 100644
--- a/docs/apache-airflow/lineage.rst
+++ b/docs/apache-airflow/lineage.rst
@@ -95,3 +95,24 @@ has outlets defined (e.g. by using ``add_outlets(..)`` or has out of the box sup
f_in > run_this | (run_this_last > outlets)
.. _precedence: https://docs.python.org/3/reference/expressions.html
+
+
+Lineage Backend
+---------------
+
+It's possible to push the lineage metrics to a custom backend by providing an instance of a LinageBackend in the config:
+
+.. code-block:: ini
+
+ [lineage]
+ backend = my.lineage.CustomBackend
+
+The backend should inherit from ``airflow.lineage.LineageBackend``.
+
+.. code-block:: python
+
+ from airflow.lineage.backend import LineageBackend
+
+ class ExampleBackend(LineageBackend):
+ def send_lineage(self, operator, inlets=None, outlets=None, context=None):
+ # Send the info to some external service
diff --git a/tests/lineage/test_lineage.py b/tests/lineage/test_lineage.py
index 350a8be..b5ebbea 100644
--- a/tests/lineage/test_lineage.py
+++ b/tests/lineage/test_lineage.py
@@ -16,16 +16,24 @@
# specific language governing permissions and limitations
# under the License.
import unittest
+from unittest import mock
-from airflow.lineage import AUTO
+from airflow.lineage import AUTO, apply_lineage, get_backend, prepare_lineage
+from airflow.lineage.backend import LineageBackend
from airflow.lineage.entities import File
from airflow.models import DAG, TaskInstance as TI
from airflow.operators.dummy import DummyOperator
from airflow.utils import timezone
+from tests.test_utils.config import conf_vars
DEFAULT_DATE = timezone.datetime(2016, 1, 1)
+class CustomLineageBackend(LineageBackend):
+ def send_lineage(self, operator, inlets=None, outlets=None, context=None):
+ pass
+
+
class TestLineage(unittest.TestCase):
def test_lineage(self):
dag = DAG(dag_id='test_prepare_lineage', start_date=DEFAULT_DATE)
@@ -111,3 +119,42 @@ class TestLineage(unittest.TestCase):
op1.pre_execute(ctx1)
assert op1.inlets[0].url == f1s.format(DEFAULT_DATE)
assert op1.outlets[0].url == f1s.format(DEFAULT_DATE)
+
+ @mock.patch("airflow.lineage.get_backend")
+ def test_lineage_is_sent_to_backend(self, mock_get_backend):
+ class TestBackend(LineageBackend):
+ def send_lineage(self, operator, inlets=None, outlets=None, context=None):
+ assert len(inlets) == 1
+ assert len(outlets) == 1
+
+ func = mock.Mock()
+ func.__name__ = 'foo'
+
+ mock_get_backend.return_value = TestBackend()
+
+ dag = DAG(dag_id='test_lineage_is_sent_to_backend', start_date=DEFAULT_DATE)
+
+ with dag:
+ op1 = DummyOperator(task_id='task1')
+
+ file1 = File("/tmp/some_file")
+
+ op1.inlets.append(file1)
+ op1.outlets.append(file1)
+
+ ctx1 = {"ti": TI(task=op1, execution_date=DEFAULT_DATE), "execution_date": DEFAULT_DATE}
+
+ prep = prepare_lineage(func)
+ prep(op1, ctx1)
+ post = apply_lineage(func)
+ post(op1, ctx1)
+
+ def test_empty_lineage_backend(self):
+ backend = get_backend()
+ assert backend is None
+
+ @conf_vars({("lineage", "backend"): "tests.lineage.test_lineage.CustomLineageBackend"})
+ def test_resolve_lineage_class(self):
+ backend = get_backend()
+ assert issubclass(backend.__class__, LineageBackend)
+ assert isinstance(backend, CustomLineageBackend)
[airflow] 04/08: Fix password masking in CLI action_logging (#15143)
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ash pushed a commit to branch v2-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit cfeeb1474e74c8ccc08ebb5c9714c8772723b092
Author: Xiaodong DENG <xd...@apache.org>
AuthorDate: Thu Apr 1 23:02:28 2021 +0200
Fix password masking in CLI action_logging (#15143)
Currently as long as argument '-p' if present, code tries to mask it.
However, '-p' may mean something else (not password), like a boolean flag. Such cases may result in exception
(cherry picked from commit 486b76438c0679682cf98cb88ed39c4b161cbcc8)
---
airflow/utils/cli.py | 20 +++++++++++---------
tests/utils/test_cli_util.py | 10 ++++++++++
2 files changed, 21 insertions(+), 9 deletions(-)
diff --git a/airflow/utils/cli.py b/airflow/utils/cli.py
index 68a0b44..fc73dfc 100644
--- a/airflow/utils/cli.py
+++ b/airflow/utils/cli.py
@@ -110,17 +110,19 @@ def _build_metrics(func_name, namespace):
"""
from airflow.models import Log
+ sub_commands_to_check = {'users', 'connections'}
sensitive_fields = {'-p', '--password', '--conn-password'}
full_command = list(sys.argv)
- for idx, command in enumerate(full_command): # pylint: disable=too-many-nested-blocks
- if command in sensitive_fields:
- # For cases when password is passed as "--password xyz" (with space between key and value)
- full_command[idx + 1] = "*" * 8
- else:
- # For cases when password is passed as "--password=xyz" (with '=' between key and value)
- for sensitive_field in sensitive_fields:
- if command.startswith(f'{sensitive_field}='):
- full_command[idx] = f'{sensitive_field}={"*" * 8}'
+ if full_command[1] in sub_commands_to_check: # pylint: disable=too-many-nested-blocks
+ for idx, command in enumerate(full_command):
+ if command in sensitive_fields:
+ # For cases when password is passed as "--password xyz" (with space between key and value)
+ full_command[idx + 1] = "*" * 8
+ else:
+ # For cases when password is passed as "--password=xyz" (with '=' between key and value)
+ for sensitive_field in sensitive_fields:
+ if command.startswith(f'{sensitive_field}='):
+ full_command[idx] = f'{sensitive_field}={"*" * 8}'
metrics = {
'sub_command': func_name,
diff --git a/tests/utils/test_cli_util.py b/tests/utils/test_cli_util.py
index c567f44..6d88f66 100644
--- a/tests/utils/test_cli_util.py
+++ b/tests/utils/test_cli_util.py
@@ -112,9 +112,19 @@ class TestCliUtil(unittest.TestCase):
"airflow connections add dsfs --conn-login asd --conn-password test --conn-type google",
"airflow connections add dsfs --conn-login asd --conn-password ******** --conn-type google",
),
+ (
+ "airflow scheduler -p",
+ "airflow scheduler -p",
+ ),
+ (
+ "airflow celery flower -p 8888",
+ "airflow celery flower -p 8888",
+ ),
]
)
def test_cli_create_user_supplied_password_is_masked(self, given_command, expected_masked_command):
+ # '-p' value which is not password, like 'airflow scheduler -p'
+ # or 'airflow celery flower -p 8888', should not be masked
args = given_command.split()
expected_command = expected_masked_command.split()
[airflow] 06/08: Change default of `[kubernetes]
enable_tcp_keepalive` to `True` (#15338)
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ash pushed a commit to branch v2-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7e28c975b66a942a7253092f2dff6619d7882f63
Author: Jed Cunningham <66...@users.noreply.github.com>
AuthorDate: Tue Apr 13 08:24:09 2021 -0600
Change default of `[kubernetes] enable_tcp_keepalive` to `True` (#15338)
We've seen instances of connection resets happening, particularly in
Azure, that are remedied by enabling tcp_keepalive. Enabling it by
default should be safe and sane regardless of where we are running.
(cherry picked from commit 6e31465a30dfd17e2e1409a81600b2e83c910036)
---
airflow/config_templates/config.yml | 2 +-
airflow/config_templates/default_airflow.cfg | 2 +-
airflow/kubernetes/kube_client.py | 8 ++++----
3 files changed, 6 insertions(+), 6 deletions(-)
diff --git a/airflow/config_templates/config.yml b/airflow/config_templates/config.yml
index 32694e4..c92acf8 100644
--- a/airflow/config_templates/config.yml
+++ b/airflow/config_templates/config.yml
@@ -2065,7 +2065,7 @@
version_added: ~
type: boolean
example: ~
- default: "False"
+ default: "True"
- name: tcp_keep_idle
description: |
When the `enable_tcp_keepalive` option is enabled, TCP probes a connection that has
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index 7685457..bc4d54a 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -1014,7 +1014,7 @@ delete_option_kwargs =
# Enables TCP keepalive mechanism. This prevents Kubernetes API requests to hang indefinitely
# when idle connection is time-outed on services like cloud load balancers or firewalls.
-enable_tcp_keepalive = False
+enable_tcp_keepalive = True
# When the `enable_tcp_keepalive` option is enabled, TCP probes a connection that has
# been idle for `tcp_keep_idle` seconds.
diff --git a/airflow/kubernetes/kube_client.py b/airflow/kubernetes/kube_client.py
index 7e8c5e8..1e65ae5 100644
--- a/airflow/kubernetes/kube_client.py
+++ b/airflow/kubernetes/kube_client.py
@@ -80,9 +80,9 @@ def _enable_tcp_keepalive() -> None:
from urllib3.connection import HTTPConnection, HTTPSConnection
- tcp_keep_idle = conf.getint('kubernetes', 'tcp_keep_idle', fallback=120)
- tcp_keep_intvl = conf.getint('kubernetes', 'tcp_keep_intvl', fallback=30)
- tcp_keep_cnt = conf.getint('kubernetes', 'tcp_keep_cnt', fallback=6)
+ tcp_keep_idle = conf.getint('kubernetes', 'tcp_keep_idle')
+ tcp_keep_intvl = conf.getint('kubernetes', 'tcp_keep_intvl')
+ tcp_keep_cnt = conf.getint('kubernetes', 'tcp_keep_cnt')
socket_options = [
(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
@@ -120,7 +120,7 @@ def get_kube_client(
if config_file is None:
config_file = conf.get('kubernetes', 'config_file', fallback=None)
- if conf.getboolean('kubernetes', 'enable_tcp_keepalive', fallback=False):
+ if conf.getboolean('kubernetes', 'enable_tcp_keepalive'):
_enable_tcp_keepalive()
client_conf = _get_kube_config(in_cluster, cluster_context, config_file)
[airflow] 07/08: Fix missing on_load trigger for folder-based
plugins (#15208)
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ash pushed a commit to branch v2-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7b9f0915144af397bd4b2e465af0875a3a6a6bc3
Author: Jed Cunningham <66...@users.noreply.github.com>
AuthorDate: Tue Apr 6 15:48:12 2021 -0600
Fix missing on_load trigger for folder-based plugins (#15208)
(cherry picked from commit 97b7780df48b412e104ff4adeecbe715264f00eb)
---
airflow/plugins_manager.py | 23 +++++++++-------
tests/plugins/test_plugin.py | 7 +++++
tests/plugins/test_plugins_manager.py | 49 +++++++++++++++++++++++++++++++++++
3 files changed, 70 insertions(+), 9 deletions(-)
diff --git a/airflow/plugins_manager.py b/airflow/plugins_manager.py
index b68dbb9..cf957ff 100644
--- a/airflow/plugins_manager.py
+++ b/airflow/plugins_manager.py
@@ -173,13 +173,23 @@ def is_valid_plugin(plugin_obj):
return False
+def register_plugin(plugin_instance):
+ """
+ Start plugin load and register it after success initialization
+
+ :param plugin_instance: subclass of AirflowPlugin
+ """
+ global plugins # pylint: disable=global-statement
+ plugin_instance.on_load()
+ plugins.append(plugin_instance)
+
+
def load_entrypoint_plugins():
"""
Load and register plugins AirflowPlugin subclasses from the entrypoints.
The entry_point group should be 'airflow.plugins'.
"""
global import_errors # pylint: disable=global-statement
- global plugins # pylint: disable=global-statement
log.debug("Loading plugins from entrypoints")
@@ -191,10 +201,8 @@ def load_entrypoint_plugins():
continue
plugin_instance = plugin_class()
- if callable(getattr(plugin_instance, 'on_load', None)):
- plugin_instance.on_load()
- plugin_instance.source = EntryPointSource(entry_point, dist)
- plugins.append(plugin_instance)
+ plugin_instance.source = EntryPointSource(entry_point, dist)
+ register_plugin(plugin_instance)
except Exception as e: # pylint: disable=broad-except
log.exception("Failed to import plugin %s", entry_point.name)
import_errors[entry_point.module] = str(e)
@@ -203,11 +211,9 @@ def load_entrypoint_plugins():
def load_plugins_from_plugin_directory():
"""Load and register Airflow Plugins from plugins directory"""
global import_errors # pylint: disable=global-statement
- global plugins # pylint: disable=global-statement
log.debug("Loading plugins from directory: %s", settings.PLUGINS_FOLDER)
for file_path in find_path_from_directory(settings.PLUGINS_FOLDER, ".airflowignore"):
-
if not os.path.isfile(file_path):
continue
mod_name, file_ext = os.path.splitext(os.path.split(file_path)[-1])
@@ -225,8 +231,7 @@ def load_plugins_from_plugin_directory():
for mod_attr_value in (m for m in mod.__dict__.values() if is_valid_plugin(m)):
plugin_instance = mod_attr_value()
plugin_instance.source = PluginsDirectorySource(file_path)
- plugins.append(plugin_instance)
-
+ register_plugin(plugin_instance)
except Exception as e: # pylint: disable=broad-except
log.exception(e)
log.error('Failed to import plugin %s', file_path)
diff --git a/tests/plugins/test_plugin.py b/tests/plugins/test_plugin.py
index d52d8e5..ca02a39 100644
--- a/tests/plugins/test_plugin.py
+++ b/tests/plugins/test_plugin.py
@@ -127,3 +127,10 @@ class MockPluginB(AirflowPlugin):
class MockPluginC(AirflowPlugin):
name = 'plugin-c'
+
+
+class AirflowTestOnLoadPlugin(AirflowPlugin):
+ name = 'preload'
+
+ def on_load(self, *args, **kwargs):
+ self.name = 'postload'
diff --git a/tests/plugins/test_plugins_manager.py b/tests/plugins/test_plugins_manager.py
index f730f17..7c4d86a 100644
--- a/tests/plugins/test_plugins_manager.py
+++ b/tests/plugins/test_plugins_manager.py
@@ -17,18 +17,33 @@
# under the License.
import importlib
import logging
+import os
import sys
+import tempfile
import unittest
from unittest import mock
+import pytest
+
from airflow.hooks.base import BaseHook
from airflow.plugins_manager import AirflowPlugin
from airflow.www import app as application
+from tests.test_utils.config import conf_vars
from tests.test_utils.mock_plugins import mock_plugin_manager
py39 = sys.version_info >= (3, 9)
importlib_metadata = 'importlib.metadata' if py39 else 'importlib_metadata'
+ON_LOAD_EXCEPTION_PLUGIN = """
+from airflow.plugins_manager import AirflowPlugin
+
+class AirflowTestOnLoadExceptionPlugin(AirflowPlugin):
+ name = 'preload'
+
+ def on_load(self, *args, **kwargs):
+ raise Exception("oops")
+"""
+
class TestPluginsRBAC(unittest.TestCase):
def setUp(self):
@@ -145,6 +160,40 @@ class TestPluginsManager:
assert caplog.records[-1].levelname == 'DEBUG'
assert caplog.records[-1].msg == 'Loading %d plugin(s) took %.2f seconds'
+ def test_loads_filesystem_plugins(self, caplog):
+ from airflow import plugins_manager
+
+ with mock.patch('airflow.plugins_manager.plugins', []):
+ plugins_manager.load_plugins_from_plugin_directory()
+
+ assert 5 == len(plugins_manager.plugins)
+ for plugin in plugins_manager.plugins:
+ if 'AirflowTestOnLoadPlugin' not in str(plugin):
+ continue
+ assert 'postload' == plugin.name
+ break
+ else:
+ pytest.fail("Wasn't able to find a registered `AirflowTestOnLoadPlugin`")
+
+ assert caplog.record_tuples == []
+
+ def test_loads_filesystem_plugins_exception(self, caplog):
+ from airflow import plugins_manager
+
+ with mock.patch('airflow.plugins_manager.plugins', []):
+ with tempfile.TemporaryDirectory() as tmpdir:
+ with open(os.path.join(tmpdir, 'testplugin.py'), "w") as f:
+ f.write(ON_LOAD_EXCEPTION_PLUGIN)
+
+ with conf_vars({('core', 'plugins_folder'): tmpdir}):
+ plugins_manager.load_plugins_from_plugin_directory()
+
+ assert plugins_manager.plugins == []
+
+ received_logs = caplog.text
+ assert 'Failed to import plugin' in received_logs
+ assert 'testplugin.py' in received_logs
+
def test_should_warning_about_incompatible_plugins(self, caplog):
class AirflowAdminViewsPlugin(AirflowPlugin):
name = "test_admin_views_plugin"
[airflow] 02/08: Add documentation create/update community
providers (#15061)
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ash pushed a commit to branch v2-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit cf3de8ff1b7c2f14deead56051f9c0253a13f00d
Author: Marcos Marx <ma...@users.noreply.github.com>
AuthorDate: Sat Apr 3 10:01:04 2021 -0300
Add documentation create/update community providers (#15061)
(cherry picked from commit 932f8c2e9360de6371031d4d71df00867a2776e6)
---
.../howto/create-update-providers.rst | 301 +++++++++++++++++++++
docs/apache-airflow-providers/index.rst | 14 +-
2 files changed, 314 insertions(+), 1 deletion(-)
diff --git a/docs/apache-airflow-providers/howto/create-update-providers.rst b/docs/apache-airflow-providers/howto/create-update-providers.rst
new file mode 100644
index 0000000..47ebb77
--- /dev/null
+++ b/docs/apache-airflow-providers/howto/create-update-providers.rst
@@ -0,0 +1,301 @@
+
+ .. 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.
+
+Community Providers
+===================
+
+.. contents:: :local:
+
+How-to creating a new community provider
+----------------------------------------
+
+This document gathers the necessary steps to create a new community provider and also guidelines for updating
+the existing ones. You should be aware that providers may have distinctions that may not be covered in
+this guide. The sequence described was designed to meet the most linear flow possible in order to develop a
+new provider.
+
+Another recommendation that will help you is to look for a provider that works similar to yours. That way it will
+help you to set up tests and other dependencies.
+
+First, you need to set up your local development environment. See `Contribution Quick Start <https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst>`_
+if you did not set up your local environment yet. We recommend using ``breeze`` to develop locally. This way you
+easily be able to have an environment more similar to the one executed by Github CI workflow.
+
+ .. code-block:: bash
+
+ ./breeze
+
+Using the code above you will set up Docker containers. These containers your local code to internal volumes.
+In this way, the changes made in your IDE are already applied to the code inside the container and tests can
+be carried out quickly.
+
+In this how-to guide our example provider name will be ``<NEW_PROVIDER>``.
+When you see this placeholder you must change for your provider name.
+
+
+Initial Code and Unit Tests
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Most likely you have developed a version of the provider using some local customization and now you need to
+transfer this code to the Airflow project. Below is described all the initial code structure that
+the provider may need. Understand that not all providers will need all the components described in this structure.
+If you still have doubts about building your provider, we recommend that you read the initial provider guide and
+open a issue on Github so the community can help you.
+
+ .. code-block:: bash
+
+ airflow/
+ ├── providers/<NEW_PROVIDER>/
+ │ ├── __init__.py
+ │ ├── example_dags/
+ │ │ ├── __init__.py
+ │ │ └── example_<NEW_PROVIDER>.py
+ │ ├── hooks/
+ │ │ ├── __init__.py
+ │ │ └── <NEW_PROVIDER>.py
+ │ ├── operators/
+ │ │ ├── __init__.py
+ │ │ └── <NEW_PROVIDER>.py
+ │ ├── sensors/
+ │ │ ├── __init__.py
+ │ │ └── <NEW_PROVIDER>.py
+ │ └── transfers/
+ │ ├── __init__.py
+ │ └── <NEW_PROVIDER>.py
+ └── tests/providers/<NEW_PROVIDER>/
+ ├── __init__.py
+ ├── hooks/
+ │ ├── __init__.py
+ │ └── test_<NEW_PROVIDER>.py
+ ├── operators/
+ │ ├── __init__.py
+ │ ├── test_<NEW_PROVIDER>.py
+ │ └── test_<NEW_PROVIDER>_system.py
+ ├── sensors/
+ │ ├── __init__.py
+ │ └── test_<NEW_PROVIDER>.py
+ └── transfers/
+ ├── __init__.py
+ └── test_<NEW_PROVIDER>.py
+
+Considering that you have already transferred your provider's code to the above structure, it will now be necessary
+to create unit tests for each component you created. The example below I have already set up an environment using
+breeze and I'll run unit tests for my Hook.
+
+ .. code-block:: bash
+
+ root@fafd8d630e46:/opt/airflow# python -m pytest tests/providers/<NEW_PROVIDER>/hook/<NEW_PROVIDER>.py
+
+Update Airflow validation tests
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+There are some tests that Airflow performs to ensure consistency that is related to the providers.
+
+ .. code-block:: bash
+
+ airflow/scripts/in_container/
+ └── run_install_and_test_provider_packages.sh
+ tests/core/
+ └── test_providers_manager.py
+
+Change expected number of providers, hooks and connections if needed in ``run_install_and_test_provider_packages.sh`` file.
+
+Add your provider information in the following variables in ``test_providers_manager.py``:
+
+- add your provider to ``ALL_PROVIDERS`` list;
+- add your provider into ``CONNECTIONS_LIST`` if your provider create a new connection type.
+
+
+Integration tests
+^^^^^^^^^^^^^^^^^
+
+See `Airflow Integration Tests <https://github.com/apache/airflow/blob/master/TESTING.rst#airflow-integration-tests>`_
+
+
+Documentation
+^^^^^^^^^^^^^
+
+An important part of building a new provider is the documentation.
+Some steps for documentation occurs automatically by ``pre-commit`` see `Installing pre-commit guide <https://github.com/apache/airflow/blob/master/CONTRIBUTORS_QUICK_START.rst#pre-commit>`_
+
+ .. code-block:: bash
+
+ airflow/
+ ├── INSTALL
+ ├── CONTRIBUTING.rst
+ ├── setup.py
+ ├── docs/
+ │ ├── spelling_wordlist.txt
+ │ ├── apache-airflow/
+ │ │ └── extra-packages-ref.rst
+ │ ├── integration-logos/<NEW_PROVIDER>/
+ │ │ └── <NEW_PROVIDER>.png
+ │ └── apache-airflow-providers-<NEW_PROVIDER>/
+ │ ├── index.rst
+ │ ├── commits.rst
+ │ ├── connections.rst
+ │ └── operators/
+ │ └── <NEW_PROVIDER>.rst
+ └── providers/
+ ├── dependencies.json
+ └── <NEW_PROVIDER>/
+ ├── provider.yaml
+ └── CHANGELOG.rst
+
+
+Files automatically updated by pre-commit:
+
+- ``airflow/providers/dependencies.json``
+- ``INSTALL``
+
+Files automatically created when the provider is released:
+
+- ``docs/apache-airflow-providers-<NEW_PROVIDER>/commits.rst``
+- ``/airflow/providers/<NEW_PROVIDER>/CHANGELOG``
+
+There is a chance that your provider's name is not a common English word.
+In this case is necessary to add it to the file ``docs/spelling_wordlist.txt``. This file begin with capitalized words and
+lowercase in the second block.
+
+ .. code-block:: bash
+
+ Namespace
+ Neo4j
+ Nextdoor
+ <NEW_PROVIDER> (new line)
+ Nones
+ NotFound
+ Nullable
+ ...
+ neo4j
+ neq
+ networkUri
+ <NEW_PROVIDER> (new line)
+ nginx
+ nobr
+ nodash
+
+Add your provider dependencies into **PROVIDER_REQUIREMENTS** variable in ``setup.py``. If your provider doesn't have
+any dependency add a empty list.
+
+ .. code-block:: python
+
+ PROVIDERS_REQUIREMENTS: Dict[str, List[str]] = {
+ ...
+ 'microsoft.winrm': winrm,
+ 'mongo': mongo,
+ 'mysql': mysql,
+ 'neo4j': neo4j,
+ '<NEW_PROVIDER>': [],
+ 'odbc': odbc,
+ ...
+ }
+
+In the ``CONTRIBUTING.rst`` adds:
+
+- your provider name in the list in the **Extras** section
+- your provider dependencies in the **Provider Packages** section table, only if your provider has external dependencies.
+
+In the ``docs/apache-airflow-providers-<NEW_PROVIDER>/connections.rst``:
+
+- add information how to configure connection for your provider.
+
+In the ``docs/apache-airflow-providers-<NEW_PROVIDER>/operators/<NEW_PROVIDER>.rst``:
+
+- add information how to use the Operator. It's important to add examples and additional information if your Operator has extra-parameters.
+
+ .. code-block:: RST
+
+ .. _howto/operator:NewProviderOperator:
+
+ NewProviderOperator
+ ===================
+
+ Use the :class:`~airflow.providers.<NEW_PROVIDER>.operators.NewProviderOperator` to do something
+ amazing with Airflow!
+
+ Using the Operator
+ ^^^^^^^^^^^^^^^^^^
+
+ The NewProviderOperator requires a ``connection_id`` and this other awesome parameter.
+ You can see an example below:
+
+ .. exampleinclude:: /../../airflow/providers/<NEW_PROVIDER>/example_dags/example_<NEW_PROVIDER>.py
+ :language: python
+ :start-after: [START howto_operator_<NEW_PROVIDER>]
+ :end-before: [END howto_operator_<NEW_PROVIDER>]
+
+
+In the ``docs/apache-airflow-providers-new_provider/index.rst``:
+
+- add all information of the purpose of your provider. It is recommended to check with another provider to help you complete this document as best as possible.
+
+In the ``airflow/providers/<NEW_PROVIDER>/provider.yaml`` add information of your provider:
+
+ .. code-block:: yaml
+
+ package-name: apache-airflow-providers-<NEW_PROVIDER>
+ name: <NEW_PROVIDER>
+ description: |
+ `<NEW_PROVIDER> <https://example.io/>`__
+ versions:
+ - 1.0.0
+
+ integrations:
+ - integration-name: <NEW_PROVIDER>
+ external-doc-url: https://www.example.io/
+ logo: /integration-logos/<NEW_PROVIDER>/<NEW_PROVIDER>.png
+ how-to-guide:
+ - /docs/apache-airflow-providers-<NEW_PROVIDER>/operators/<NEW_PROVIDER>.rst
+ tags: [service]
+
+ operators:
+ - integration-name: <NEW_PROVIDER>
+ python-modules:
+ - airflow.providers.<NEW_PROVIDER>.operators.<NEW_PROVIDER>
+
+ hooks:
+ - integration-name: <NEW_PROVIDER>
+ python-modules:
+ - airflow.providers.<NEW_PROVIDER>.hooks.<NEW_PROVIDER>
+
+ sensors:
+ - integration-name: <NEW_PROVIDER>
+ python-modules:
+ - airflow.providers.<NEW_PROVIDER>.sensors.<NEW_PROVIDER>
+
+ hook-class-names:
+ - airflow.providers.<NEW_PROVIDER>.hooks.<NEW_PROVIDER>.NewProviderHook
+
+You only need to add ``hook-class-names`` in case you have some hooks that have customized UI behavior.
+For more information see `Custom connection types <http://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html#custom-connection-types>`_
+
+
+After changing and creating these files you can build the documentation locally. The two commands below will
+serve to accomplish this. The first will build your provider's documentation. The second will ensure that the
+main Airflow documentation that involves some steps with the providers is also working.
+
+ .. code-block:: bash
+
+ ./breeze build-docs -- --package-filter apache-airflow-providers-<NEW_PROVIDER>
+ ./breeze build-docs -- --package-filter apache-airflow
+
+How-to Update a community provider
+----------------------------------
+
+See `Provider packages versioning <https://github.com/apache/airflow/blob/master/dev/README_RELEASE_PROVIDER_PACKAGES.md#provider-packages-versioning>`_
diff --git a/docs/apache-airflow-providers/index.rst b/docs/apache-airflow-providers/index.rst
index 81e02fa..43c1687 100644
--- a/docs/apache-airflow-providers/index.rst
+++ b/docs/apache-airflow-providers/index.rst
@@ -54,6 +54,11 @@ provider packages are automatically documented in the release notes of every pro
Those are the same providers as for 2.0 but automatically back-ported to work for Airflow 1.10. The
last release of backport providers was done on March 17, 2021.
+Creating and maintaining community providers
+""""""""""""""""""""""""""""""""""""""""""""
+
+See :doc:`howto/create-update-providers` for more information.
+
Provider packages functionality
'''''''''''''''''''''''''''''''
@@ -242,7 +247,7 @@ Example ``myproviderpackage/somemodule.py``:
**How do provider packages work under the hood?**
-When running airflow with your provider package, there will be (at least) three components to your airflow installation:
+When running Airflow with your provider package, there will be (at least) three components to your airflow installation:
* The installation itself (for example, a ``venv`` where you installed airflow with ``pip install apache-airflow``)
together with the related files (e.g. ``dags`` folder)
@@ -296,6 +301,12 @@ The Community only accepts providers that are generic enough, are well documente
and with capabilities of being tested by people in the community. So we might not always be in the
position to accept such contributions.
+
+After you think that your provider matches the expected values above, you can read
+:doc:`howto/create-update-providers` to check all prerequisites for a new
+community Provider and discuss it at the `Devlist <http://airflow.apache.org/community/>`_.
+
+
However, in case you have your own, specific provider, which you can maintain on your own or by your
team, you are free to publish the providers in whatever form you find appropriate. The custom and
community-managed providers have exactly the same capabilities.
@@ -323,3 +334,4 @@ Content
Packages <packages-ref>
Operators and hooks <operators-and-hooks-ref/index>
+ Howto create and update community providers <howto/create-update-providers>
[airflow] 08/08: Import Connection lazily in hooks to avoid cycles
(#15361)
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ash pushed a commit to branch v2-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit ef876cfe33e723110b5fb9fb527d874eb757ab32
Author: Tzu-ping Chung <tp...@astronomer.io>
AuthorDate: Wed Apr 14 21:33:00 2021 +0800
Import Connection lazily in hooks to avoid cycles (#15361)
The current implementation imports Connection on import time, which
causes a circular import when a model class needs to reference a hook
class.
By applying this fix, the airflow.hooks package is completely decoupled
with airflow.models on import time, so model code can reference hooks.
Hooks, on the other hand, generally don't reference model classes.
Fix #15325.
(cherry picked from commit 75603160848e4199ed368809dfd441dcc5ddbd82)
---
airflow/hooks/base.py | 12 ++++++++----
1 file changed, 8 insertions(+), 4 deletions(-)
diff --git a/airflow/hooks/base.py b/airflow/hooks/base.py
index b3c0c11..dee76dc 100644
--- a/airflow/hooks/base.py
+++ b/airflow/hooks/base.py
@@ -18,12 +18,14 @@
"""Base class for all hooks"""
import logging
import warnings
-from typing import Any, Dict, List
+from typing import TYPE_CHECKING, Any, Dict, List
-from airflow.models.connection import Connection
from airflow.typing_compat import Protocol
from airflow.utils.log.logging_mixin import LoggingMixin
+if TYPE_CHECKING:
+ from airflow.models.connection import Connection # Avoid circular imports.
+
log = logging.getLogger(__name__)
@@ -37,7 +39,7 @@ class BaseHook(LoggingMixin):
"""
@classmethod
- def get_connections(cls, conn_id: str) -> List[Connection]:
+ def get_connections(cls, conn_id: str) -> List["Connection"]:
"""
Get all connections as an iterable, given the connection id.
@@ -53,13 +55,15 @@ class BaseHook(LoggingMixin):
return [cls.get_connection(conn_id)]
@classmethod
- def get_connection(cls, conn_id: str) -> Connection:
+ def get_connection(cls, conn_id: str) -> "Connection":
"""
Get connection, given connection id.
:param conn_id: connection id
:return: connection
"""
+ from airflow.models.connection import Connection
+
conn = Connection.get_connection_from_secrets(conn_id)
if conn.host:
log.info(