Deprecate *args and **kwargs in BaseOperator

BaseOperator silently accepts any arguments. This deprecates the
behavior with a warning that says it will be forbidden in Airflow 2.0.

This PR also turns on DeprecationWarnings by default, which in turn
revealed that inspect.getargspec is deprecated. Here it is replaced by
`inspect.signature` (Python 3) or `funcsigs.signature` (Python 2).

Lastly, this brought to attention that example_http_operator was
passing an illegal argument.

Author: jlowin

UPDATING.md

@@ -1,15 +1,24 @@
 # Updating Airflow
 
-This file aims to document the backwards-incompatible changes in Airflow and
-assist people with migrating to a new version.
+This file documents any backwards-incompatible changes in Airflow and
+assists people when migrating to a new version.
 
-## 1.7 to 1.8
 
-### DAGs now don't start automatically when created
+## Airflow 1.8
 
-To retain the old behavior, add this to your configuration:
+### Changes to Behavior
+
+#### New DAGs are paused by default
+
+Previously, new DAGs would be scheduled immediately. To retain the old behavior, add this to airflow.cfg:
 
 ```
+[core]
 dags_are_paused_at_creation = False
 ```
 
+### Deprecated Features
+These features are marked for deprecation. They may still work (and raise a `DeprecationWarning`), but are no longer supported and will be removed entirely in Airflow 2.0
+
+#### Operators no longer accept arbitrary arguments
+Previously, `Operator.__init__()` accepted any arguments (either positional `*args` or keyword `**kwargs`) without complaint. Now, invalid arguments will be rejected.

airflow/configuration.py

@@ -8,6 +8,7 @@
 import logging
 import os
 import subprocess
+import warnings
 
 from future import standard_library
 standard_library.install_aliases()
@@ -16,6 +17,9 @@
 from collections import OrderedDict
 from configparser import ConfigParser
 
+# show DeprecationWarning and PendingDeprecationWarning
+warnings.simplefilter('default', DeprecationWarning)
+warnings.simplefilter('default', PendingDeprecationWarning)
 
 class AirflowConfigException(Exception):
     pass

airflow/example_dags/example_http_operator.py

@@ -66,7 +66,7 @@
 
 sensor = HttpSensor(
     task_id='http_sensor_check',
-    conn_id='http_default',
+    http_conn_id='http_default',
     endpoint='',
     params={},
     response_check=lambda response: True if "Google" in response.content else False,

airflow/models.py

@@ -37,6 +37,7 @@
 import socket
 import sys
 import traceback
+import warnings
 from urllib.parse import urlparse
 
 from sqlalchemy import (
@@ -1597,6 +1598,17 @@ def __init__(
             *args,
             **kwargs):
 
+        if args or kwargs:
+            # TODO remove *args and **kwargs in Airflow 2.0
+            warnings.warn(
+                'Invalid arguments were passed to {c}. Support for '
+                'passing such arguments will be dropped in Airflow 2.0.'
+                'Invalid arguments were:'
+                '\n*args: {a}\n**kwargs: {k}'.format(
+                    c=self.__class__.__name__, a=args, k=kwargs),
+                category=PendingDeprecationWarning
+            )
+
         validate_key(task_id)
         self.dag_id = dag.dag_id if dag else 'adhoc_' + owner
         self.task_id = task_id

airflow/utils/decorators.py

@@ -12,9 +12,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-import inspect
 import os
 
+# inspect.signature is only available in Python 3. funcsigs.signature is
+# a backport.
+try:
+    import inspect
+    signature = inspect.signature
+except AttributeError:
+    import funcsigs
+    signature = funcsigs.signature
+
 from copy import copy
 from functools import wraps
 
@@ -57,12 +65,14 @@ def wrapper(*args, **kwargs):
 
         dag_args.update(default_args)
         default_args = dag_args
-        arg_spec = inspect.getargspec(func)
-        num_defaults = len(arg_spec.defaults) if arg_spec.defaults else 0
-        non_optional_args = arg_spec.args[:-num_defaults]
-        if 'self' in non_optional_args:
-            non_optional_args.remove('self')
-        for arg in func.__code__.co_varnames:
+
+        sig = signature(func)
+        non_optional_args = [
+            name for (name, param) in sig.parameters.items()
+            if param.default == param.empty and
+            param.name != 'self' and
+            param.kind not in (param.VAR_POSITIONAL, param.VAR_KEYWORD)]
+        for arg in sig.parameters:
             if arg in default_args and arg not in kwargs:
                 kwargs[arg] = default_args[arg]
         missing_args = list(set(non_optional_args) - set(kwargs))

setup.py

@@ -122,6 +122,7 @@ def run(self):
         'flask-cache>=0.13.1, <0.14',
         'flask-login==0.2.11',
         'future>=0.15.0, <0.16',
+        'funcsigs>=0.4, <1'
         'gunicorn>=19.3.0, <19.4.0',  # 19.4.? seemed to have issues
         'jinja2>=2.7.3, <3.0',
         'markdown>=2.5.2, <3.0',

tests/core.py

@@ -12,6 +12,7 @@
 from email.mime.application import MIMEApplication
 import signal
 from time import sleep
+import warnings
 
 from dateutil.relativedelta import relativedelta
 
@@ -320,6 +321,22 @@ def test_clear_api(self):
         ti = models.TaskInstance(task=task, execution_date=DEFAULT_DATE)
         ti.are_dependents_done()
 
+    def test_illegal_args(self):
+        """
+        Tests that Operators reject illegal arguments
+        """
+        with warnings.catch_warnings(record=True) as w:
+            t = operators.BashOperator(
+                task_id='test_illegal_args',
+                bash_command='echo success',
+                dag=self.dag,
+                illegal_argument_1234='hello?')
+            self.assertTrue(
+                issubclass(w[0].category, PendingDeprecationWarning))
+            self.assertIn(
+                'Invalid arguments were passed to BashOperator.',
+                w[0].message.args[0])
+
     def test_bash_operator(self):
         t = operators.BashOperator(
             task_id='time_sensor_check',