Add base OpenLineage provider implementation

.github/ISSUE_TEMPLATE/airflow_providers_bug_report.yml

@@ -75,6 +75,7 @@ body:
         - neo4j
         - odbc
         - openfaas
+        - openlineage
         - opsgenie
         - oracle
         - pagerduty

.github/workflows/ci.yml

@@ -753,6 +753,7 @@ jobs:
         run: |
           rm -vf dist/apache_airflow-*.whl
           # remove the provider packages that are not compatible with 2.3
+          rm -vf dist/apache_airflow_providers_openlineage*.whl
           # rm -vf dist/apache_airflow_providers_docker*.whl
           # pip download --no-deps --dest dist apache-airflow-providers-docker==3.1.0
         if: matrix.package-format == 'wheel'

CONTRIBUTING.rst

@@ -618,10 +618,10 @@ devel_ci, devel_hadoop, dingding, discord, doc, doc_gen, docker, druid, elastics
 facebook, ftp, gcp, gcp_api, github, github_enterprise, google, google_auth, grpc, hashicorp, hdfs,
 hive, http, imap, influxdb, jdbc, jenkins, kerberos, kubernetes, ldap, leveldb, microsoft.azure,
 microsoft.mssql, microsoft.psrp, microsoft.winrm, mongo, mssql, mysql, neo4j, odbc, openfaas,
-opsgenie, oracle, otel, pagerduty, pandas, papermill, password, pinot, plexus, postgres, presto,
-qds, qubole, rabbitmq, redis, s3, salesforce, samba, segment, sendgrid, sentry, sftp, singularity,
-slack, smtp, snowflake, spark, sqlite, ssh, statsd, tableau, tabular, telegram, trino, vertica,
-virtualenv, webhdfs, winrm, yandex, zendesk
+openlineage, opsgenie, oracle, otel, pagerduty, pandas, papermill, password, pinot, plexus,
+postgres, presto, qds, qubole, rabbitmq, redis, s3, salesforce, samba, segment, sendgrid, sentry,
+sftp, singularity, slack, smtp, snowflake, spark, sqlite, ssh, statsd, tableau, tabular, telegram,
+trino, vertica, virtualenv, webhdfs, winrm, yandex, zendesk
   .. END EXTRAS HERE
 
 Provider packages

INSTALL

@@ -103,10 +103,10 @@ devel_ci, devel_hadoop, dingding, discord, doc, doc_gen, docker, druid, elastics
 facebook, ftp, gcp, gcp_api, github, github_enterprise, google, google_auth, grpc, hashicorp, hdfs,
 hive, http, imap, influxdb, jdbc, jenkins, kerberos, kubernetes, ldap, leveldb, microsoft.azure,
 microsoft.mssql, microsoft.psrp, microsoft.winrm, mongo, mssql, mysql, neo4j, odbc, openfaas,
-opsgenie, oracle, otel, pagerduty, pandas, papermill, password, pinot, plexus, postgres, presto,
-qds, qubole, rabbitmq, redis, s3, salesforce, samba, segment, sendgrid, sentry, sftp, singularity,
-slack, smtp, snowflake, spark, sqlite, ssh, statsd, tableau, tabular, telegram, trino, vertica,
-virtualenv, webhdfs, winrm, yandex, zendesk
+openlineage, opsgenie, oracle, otel, pagerduty, pandas, papermill, password, pinot, plexus,
+postgres, presto, qds, qubole, rabbitmq, redis, s3, salesforce, samba, segment, sendgrid, sentry,
+sftp, singularity, slack, smtp, snowflake, spark, sqlite, ssh, statsd, tableau, tabular, telegram,
+trino, vertica, virtualenv, webhdfs, winrm, yandex, zendesk
 # END EXTRAS HERE
 
 # For installing Airflow in development environments - see CONTRIBUTING.rst

airflow/plugins_manager.py

@@ -164,7 +164,7 @@ class AirflowPlugin:
     # A list of timetable classes that can be used for DAG scheduling.
     timetables: list[type[Timetable]] = []
 
-    listeners: list[ModuleType] = []
+    listeners: list[ModuleType | object] = []
 
     @classmethod
     def validate(cls):

airflow/providers/openlineage/CHANGELOG.rst

@@ -0,0 +1,25 @@
+ .. 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.
+
+
+Changelog
+---------
+
+1.0.0
+.....
+
+Initial version of the provider.

airflow/providers/openlineage/__init__.py

@@ -0,0 +1,38 @@
+# 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.
+from __future__ import annotations
+
+__all__ = ["version"]
+
+try:
+    import importlib_metadata as metadata
+except ImportError:
+    from importlib import metadata  # type: ignore[no-redef]
+
+try:
+    version = metadata.version("apache-airflow-providers-openlineage")
+except metadata.PackageNotFoundError:
+    import logging
+
+    log = logging.getLogger(__name__)
+    log.warning("Package metadata could not be found. Overriding it with version found in setup.py")
+    # TODO: What should be a proper fallback?
+    # If hardcoded version from provider version
+    # there's no point to use metadata above
+    version = "1.0.0.dev"
+
+del metadata

airflow/providers/openlineage/extractors/__init__.py

@@ -0,0 +1,23 @@
+# 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.
+
+from __future__ import annotations
+
+from airflow.providers.openlineage.extractors.base import BaseExtractor, OperatorLineage
+from airflow.providers.openlineage.extractors.manager import ExtractorManager
+
+__all__ = ["BaseExtractor", "OperatorLineage", "ExtractorManager"]

airflow/providers/openlineage/extractors/base.py

@@ -0,0 +1,102 @@
+# 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.
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from attrs import Factory, define
+
+from airflow.utils.log.logging_mixin import LoggingMixin
+from openlineage.client.facet import BaseFacet
+from openlineage.client.run import Dataset
+
+
+@define
+class OperatorLineage:
+    """Structure returned from lineage extraction."""
+
+    inputs: list[Dataset] = Factory(list)
+    outputs: list[Dataset] = Factory(list)
+    run_facets: dict[str, BaseFacet] = Factory(dict)
+    job_facets: dict[str, BaseFacet] = Factory(dict)
+
+
+class BaseExtractor(ABC, LoggingMixin):
+    """
+    Abstract base extractor class.
+
+    This is used mostly to maintain support for custom extractors.
+    """
+
+    _allowed_query_params: list[str] = []
+
+    def __init__(self, operator):  # type: ignore
+        super().__init__()
+        self.operator = operator
+
+    @classmethod
+    @abstractmethod
+    def get_operator_classnames(cls) -> list[str]:
+        """
+        Implement this method returning list of operators that extractor works for.
+        There are operators which work very similarly and one extractor can cover both.
+        """
+        raise NotImplementedError()
+
+    def validate(self):
+        assert self.operator.task_type in self.get_operator_classnames()
+
+    @abstractmethod
+    def extract(self) -> OperatorLineage | None:
+        pass
+
+    def extract_on_complete(self, task_instance) -> OperatorLineage | None:
+        return self.extract()
+
+
+class DefaultExtractor(BaseExtractor):
+    """Extractor that uses `get_openlineage_facets_on_start/complete/failure` methods."""
+
+    @classmethod
+    def get_operator_classnames(cls) -> list[str]:
+        """
+        Default extractor is chosen not on the classname basis, but
+        by existence of get_openlineage_facets method on operator
+        """
+        return []
+
+    def extract(self) -> OperatorLineage | None:
+        try:
+            return self._get_openlineage_facets(self.operator.get_openlineage_facets_on_start)  # type: ignore
+        except AttributeError:
+            return None
+
+    def extract_on_complete(self, task_instance) -> OperatorLineage | None:
+        on_complete = getattr(self.operator, "get_openlineage_facets_on_complete", None)
+        if on_complete and callable(on_complete):
+            return self._get_openlineage_facets(on_complete, task_instance)
+        return self.extract()
+
+    def _get_openlineage_facets(self, get_facets_method, *args) -> OperatorLineage | None:
+        facets: OperatorLineage = get_facets_method(*args)
+        return OperatorLineage(
+            inputs=facets.inputs,
+            outputs=facets.outputs,
+            run_facets=facets.run_facets,
+            job_facets=facets.job_facets,
+        )