Drop redundant project id null checks.

UPDATING.md

@@ -62,6 +62,35 @@ https://developers.google.com/style/inclusive-documentation
 
 -->
 
+### Added mypy plugin to preserve types of decorated functions
+
+Mypy currently doesn't support precise type information for decorated
+functions; see https://github.com/python/mypy/issues/3157 for details.
+To preserve precise type definitions for decorated functions, we now
+include a mypy plugin to preserve precise type definitions for decorated
+functions. To use the plugin, update your setup.cfg:
+
+```
+[mypy]
+plugins =
+  airflow.mypy.plugin.decorators
+```
+
+### Use project_id argument consistently across GCP hooks and operators
+
+- Changed order of arguments in DataflowHook.start_python_dataflow. Uses
+    with positional arguments may break.
+- Changed order of arguments in DataflowHook.is_job_dataflow_running. Uses
+    with positional arguments may break.
+- Changed order of arguments in DataflowHook.cancel_job. Uses
+    with positional arguments may break.
+- Added optional project_id argument to DataflowCreateJavaJobOperator
+    constructor.
+- Added optional project_id argument to DataflowTemplatedJobStartOperator
+    constructor.
+- Added optional project_id argument to DataflowCreatePythonJobOperator
+    constructor.
+
 ### Rename pool statsd metrics
 
 Used slot has been renamed to running slot to make the name self-explanatory

airflow/mypy/__init__.py

@@ -0,0 +1,17 @@
+#
+# 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.

airflow/mypy/plugin/__init__.py

@@ -0,0 +1,17 @@
+#
+# 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.

airflow/mypy/plugin/decorators.py

@@ -0,0 +1,79 @@
+#
+# 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.
+
+import copy
+import functools
+from typing import List
+
+from mypy.nodes import ARG_NAMED_OPT  # pylint: disable=no-name-in-module
+from mypy.plugin import FunctionContext, Plugin  # pylint: disable=no-name-in-module
+from mypy.types import CallableType, NoneType, UnionType  # pylint: disable=no-name-in-module
+
+TYPED_DECORATORS = {
+    "fallback_to_default_project_id of GoogleBaseHook": ["project_id"],
+    "airflow.providers.google.cloud.hooks.dataflow._fallback_to_project_id_from_variables": ["project_id"],
+    "provide_gcp_credential_file of GoogleBaseHook": [],
+}
+
+
+class TypedDecoratorPlugin(Plugin):
+    """Mypy plugin for typed decorators."""
+    def get_function_hook(self, fullname: str):
+        """Check for known typed decorators by name."""
+        if fullname in TYPED_DECORATORS:
+            return functools.partial(
+                _analyze_decorator,
+                provided_arguments=TYPED_DECORATORS[fullname],
+            )
+        return None
+
+
+def _analyze_decorator(function_ctx: FunctionContext, provided_arguments: List[str]):
+    if not isinstance(function_ctx.arg_types[0][0], CallableType):
+        return function_ctx.default_return_type
+    if not isinstance(function_ctx.default_return_type, CallableType):
+        return function_ctx.default_return_type
+    return _change_decorator_function_type(
+        function_ctx.arg_types[0][0],
+        function_ctx.default_return_type,
+        provided_arguments,
+    )
+
+
+def _change_decorator_function_type(
+    decorated: CallableType,
+    decorator: CallableType,
+    provided_arguments: List[str],
+) -> CallableType:
+    decorator.arg_kinds = decorated.arg_kinds
+    decorator.arg_names = decorated.arg_names
+
+    # Mark provided arguments as optional
+    decorator.arg_types = copy.copy(decorated.arg_types)
+    for argument in provided_arguments:
+        index = decorated.arg_names.index(argument)
+        decorated_type = decorated.arg_types[index]
+        decorator.arg_types[index] = UnionType.make_union([decorated_type, NoneType()])
+        decorated.arg_kinds[index] = ARG_NAMED_OPT
+
+    return decorator
+
+
+def plugin(version: str):  # pylint: disable=unused-argument
+    """Mypy plugin entrypoint."""
+    return TypedDecoratorPlugin

airflow/providers/google/cloud/example_dags/example_automl_tables.py

@@ -139,7 +139,6 @@ def get_target_column_spec(columns_specs: List[Dict], column_name: str) -> str:
         task_id="update_dataset_task",
         dataset=update,
         location=GCP_AUTOML_LOCATION,
-        project_id=GCP_PROJECT_ID,
     )
     # [END howto_operator_automl_update_dataset]