diff --git a/UPDATING.md b/UPDATING.md index a2662551cef26..7b0bc97aa0902 100644 --- a/UPDATING.md +++ b/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. diff --git a/airflow/configuration.py b/airflow/configuration.py index 64f31262e5ae3..dd8b633b92d84 100644 --- a/airflow/configuration.py +++ b/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 diff --git a/airflow/example_dags/example_http_operator.py b/airflow/example_dags/example_http_operator.py index 2e20516ffad95..bbfc17afe58c8 100644 --- a/airflow/example_dags/example_http_operator.py +++ b/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, diff --git a/airflow/models.py b/airflow/models.py index fd1349ecf5860..2cc51cc7b6705 100644 --- a/airflow/models.py +++ b/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 diff --git a/airflow/utils/decorators.py b/airflow/utils/decorators.py index 5568559699bb9..bc4678cc48db0 100644 --- a/airflow/utils/decorators.py +++ b/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)) diff --git a/setup.py b/setup.py index 924aa3245c394..4f708416435fa 100644 --- a/setup.py +++ b/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', diff --git a/tests/core.py b/tests/core.py index 484282fa964ad..373d35ebf075c 100644 --- a/tests/core.py +++ b/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',