apache · kaxil · Sep 22, 2020 · Sep 20, 2020 · Sep 20, 2020 · Sep 21, 2020
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -200,7 +200,7 @@ repos:
         name: Run pydocstyle
         args:
           - --convention=pep257
-          - --add-ignore=D100,D102,D104,D105,D107,D200,D202,D204,D205,D400,D401
+          - --add-ignore=D100,D102,D104,D105,D107,D200,D204,D205,D400,D401
         exclude: ^tests/.*\.py$|^scripts/.*\.py$|^dev|^backport_packages|^kubernetes_tests
   - repo: local
     hooks:

diff --git a/airflow/api/auth/backend/kerberos_auth.py b/airflow/api/auth/backend/kerberos_auth.py
@@ -71,7 +71,6 @@ def __init__(self):
 
 def init_app(app):
     """Initializes application with kerberos"""
-
     hostname = app.config.get('SERVER_NAME')
     if not hostname:
         hostname = getfqdn()

diff --git a/airflow/api/common/experimental/get_dag_run_state.py b/airflow/api/common/experimental/get_dag_run_state.py
@@ -29,7 +29,6 @@ def get_dag_run_state(dag_id: str, execution_date: datetime) -> Dict[str, str]:
     :param execution_date: execution date
     :return: Dictionary storing state of the object
     """
-
     dag = check_and_get_dag(dag_id=dag_id)
 
     dagrun = check_and_get_dagrun(dag, execution_date)

diff --git a/airflow/api_connexion/endpoints/dag_run_endpoint.py b/airflow/api_connexion/endpoints/dag_run_endpoint.py
@@ -87,7 +87,6 @@ def get_dag_runs(
     """
     Get all DAG Runs.
     """
-
     query = session.query(DagRun)
 
     #  This endpoint allows specifying ~ as the dag_id to retrieve DAG Runs for all DAGs.

diff --git a/airflow/api_connexion/endpoints/event_log_endpoint.py b/airflow/api_connexion/endpoints/event_log_endpoint.py
@@ -49,7 +49,6 @@ def get_event_logs(session, limit, offset=None):
     """
     Get all log entries from event log
     """
-
     total_entries = session.query(func.count(Log.id)).scalar()
     event_logs = session.query(Log).order_by(Log.id).offset(offset).limit(limit).all()
     return event_log_collection_schema.dump(

diff --git a/airflow/api_connexion/endpoints/import_error_endpoint.py b/airflow/api_connexion/endpoints/import_error_endpoint.py
@@ -52,7 +52,6 @@ def get_import_errors(session, limit, offset=None):
     """
     Get all import errors
     """
-
     total_entries = session.query(func.count(ImportError.id)).scalar()
     import_errors = session.query(ImportError).order_by(ImportError.id).offset(offset).limit(limit).all()
     return import_error_collection_schema.dump(

diff --git a/airflow/api_connexion/endpoints/pool_endpoint.py b/airflow/api_connexion/endpoints/pool_endpoint.py
@@ -60,7 +60,6 @@ def get_pools(session, limit, offset=None):
     """
     Get all pools
     """
-
     total_entries = session.query(func.count(Pool.id)).scalar()
     pools = session.query(Pool).order_by(Pool.id).offset(offset).limit(limit).all()
     return pool_collection_schema.dump(PoolCollection(pools=pools, total_entries=total_entries))

diff --git a/airflow/api_connexion/endpoints/xcom_endpoint.py b/airflow/api_connexion/endpoints/xcom_endpoint.py
@@ -47,7 +47,6 @@ def get_xcom_entries(
     """
     Get all XCom values
     """
-
     query = session.query(XCom)
     if dag_id != '~':
         query = query.filter(XCom.dag_id == dag_id)

diff --git a/airflow/api_connexion/schemas/common_schema.py b/airflow/api_connexion/schemas/common_schema.py
@@ -45,7 +45,6 @@ class TimeDeltaSchema(Schema):
     @marshmallow.post_load
     def make_time_delta(self, data, **kwargs):
         """Create time delta based on data"""
-
         if "objectType" in data:
             del data["objectType"]
         return datetime.timedelta(**data)
@@ -74,7 +73,6 @@ class RelativeDeltaSchema(Schema):
     @marshmallow.post_load
     def make_relative_delta(self, data, **kwargs):
         """Create relative delta based on data"""
-
         if "objectType" in data:
             del data["objectType"]
 

diff --git a/airflow/api_connexion/schemas/dag_schema.py b/airflow/api_connexion/schemas/dag_schema.py
@@ -56,7 +56,6 @@ class Meta:
     @staticmethod
     def get_owners(obj: DagModel):
         """Convert owners attribute to DAG representation"""
-
         if not getattr(obj, 'owners', None):
             return []
         return obj.owners.split(",")

diff --git a/airflow/cli/commands/dag_command.py b/airflow/cli/commands/dag_command.py
@@ -383,7 +383,6 @@ def dag_list_dag_runs(args, dag=None):
 @cli_utils.action_logging
 def generate_pod_yaml(args):
     """Generates yaml files for each task in the DAG. Used for testing output of KubernetesExecutor"""
-
     from kubernetes.client.api_client import ApiClient
 
     from airflow.executors.kubernetes_executor import AirflowKubernetesScheduler, KubeConfig

diff --git a/airflow/cli/commands/task_command.py b/airflow/cli/commands/task_command.py
@@ -143,7 +143,6 @@ def _run_raw_task(args, ti):
 @cli_utils.action_logging
 def task_run(args, dag=None):
     """Runs a single task instance"""
-
     # Load custom airflow config
     if args.cfg_path:
         with open(args.cfg_path, 'r') as conf_file:
@@ -289,7 +288,6 @@ def _guess_debugger():
     * `ipdb <https://github.com/gotcha/ipdb>`__
     * `pdb <https://docs.python.org/3/library/pdb.html>`__
     """
-
     for mod in SUPPORTED_DEBUGGER_MODULES:
         try:
             return importlib.import_module(mod)

diff --git a/airflow/configuration.py b/airflow/configuration.py
@@ -228,7 +228,6 @@ def _validate_config_dependencies(self):
         Validate that config values aren't invalid given other config values
         or system-level limitations and requirements.
         """
-
         if (
                 self.get("core", "executor") not in ('DebugExecutor', 'SequentialExecutor') and
                 "sqlite" in self.get('core', 'sql_alchemy_conn')):

diff --git a/airflow/executors/celery_executor.py b/airflow/executors/celery_executor.py
@@ -282,7 +282,6 @@ def _check_for_stalled_adopted_tasks(self):
 
     def update_all_task_states(self) -> None:
         """Updates states of the tasks."""
-
         self.log.debug("Inquiring about %s celery task(s)", len(self.tasks))
         state_and_info_by_celery_task_id = self.bulk_state_fetcher.get_many(self.tasks.values())
 
@@ -401,7 +400,6 @@ def fetch_celery_task_state(async_result: AsyncResult) -> \
         of the task
     :rtype: tuple[str, str, str]
     """
-
     try:
         with timeout(seconds=OPERATION_TIMEOUT):
             # Accessing state property of celery task will make actual network request

diff --git a/airflow/hooks/dbapi_hook.py b/airflow/hooks/dbapi_hook.py
@@ -220,7 +220,6 @@ def get_autocommit(self, conn):
         :return: connection autocommit setting.
         :rtype: bool
         """
-
         return getattr(conn, 'autocommit', False) and self.supports_autocommit
 
     def get_cursor(self):
@@ -320,7 +319,6 @@ def _serialize_cell(cell, conn=None):  # pylint: disable=unused-argument
         :return: The serialized cell
         :rtype: str
         """
-
         if cell is None:
             return None
         if isinstance(cell, datetime):

diff --git a/airflow/jobs/backfill_job.py b/airflow/jobs/backfill_job.py
@@ -410,7 +410,6 @@ def _process_backfill_task_instances(self,  # pylint: disable=too-many-statement
         :return: the list of execution_dates for the finished dag runs
         :rtype: list
         """
-
         executed_run_dates = []
 
         while ((len(ti_status.to_run) > 0 or len(ti_status.running) > 0) and

diff --git a/airflow/jobs/local_task_job.py b/airflow/jobs/local_task_job.py
@@ -129,7 +129,6 @@ def on_kill(self):
     @provide_session
     def heartbeat_callback(self, session=None):
         """Self destruct task if state has been moved away from running externally"""
-
         if self.terminating:
             # ensure termination if processes are created later
             self.task_runner.terminate()

diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
@@ -691,7 +691,6 @@ def _process_task_instances(
         active DAG runs and adding task instances that should run to the
         queue.
         """
-
         # update the state of the previously active dag runs
         active_dag_runs = 0
         task_instances_list = []

diff --git a/airflow/kubernetes/kube_client.py b/airflow/kubernetes/kube_client.py
@@ -80,7 +80,6 @@ def get_kube_client(in_cluster: bool = conf.getboolean('kubernetes', 'in_cluster
     :return kubernetes client
     :rtype client.CoreV1Api
     """
-
     if not has_kubernetes:
         raise _import_err
 

diff --git a/airflow/kubernetes/pod_generator.py b/airflow/kubernetes/pod_generator.py
@@ -177,7 +177,6 @@ def add_xcom_sidecar(pod: k8s.V1Pod) -> k8s.V1Pod:
     @staticmethod
     def from_obj(obj) -> Optional[Union[dict, k8s.V1Pod]]:
         """Converts to pod from obj"""
-
         if obj is None:
             return None
 

diff --git a/airflow/macros/__init__.py b/airflow/macros/__init__.py
@@ -40,7 +40,6 @@ def ds_add(ds, days):
     >>> ds_add('2015-01-06', -5)
     '2015-01-01'
     """
-
     ds = datetime.strptime(ds, '%Y-%m-%d')
     if days:
         ds = ds + timedelta(days)

diff --git a/airflow/migrations/versions/6e96a59344a4_make_taskinstance_pool_not_nullable.py b/airflow/migrations/versions/6e96a59344a4_make_taskinstance_pool_not_nullable.py
@@ -112,7 +112,6 @@ def downgrade():
     """
     Make TaskInstance.pool field nullable.
     """
-
     conn = op.get_bind()
     if conn.dialect.name == "mssql":
         op.drop_index('ti_pool', table_name='task_instance')

diff --git a/airflow/migrations/versions/8f966b9c467a_set_conn_type_as_non_nullable.py b/airflow/migrations/versions/8f966b9c467a_set_conn_type_as_non_nullable.py
@@ -37,7 +37,6 @@
 
 def upgrade():
     """Apply Set conn_type as non-nullable"""
-
     Base = declarative_base()
 
     class Connection(Base):

diff --git a/airflow/models/baseoperator.py b/airflow/models/baseoperator.py
@@ -733,7 +733,6 @@ def priority_weight_total(self) -> int:
     @cached_property
     def operator_extra_link_dict(self) -> Dict[str, Any]:
         """Returns dictionary of all extra links for the operator"""
-
         op_extra_links_from_plugin: Dict[str, Any] = {}
         from airflow import plugins_manager
         plugins_manager.initialize_extra_operators_links_plugins()
@@ -830,7 +829,6 @@ def render_template_fields(self, context: Dict, jinja_env: Optional[jinja2.Envir
         :param jinja_env: Jinja environment
         :type jinja_env: jinja2.Environment
         """
-
         if not jinja_env:
             jinja_env = self.get_template_env()
 
@@ -865,7 +863,6 @@ def render_template(      # pylint: disable=too-many-return-statements
         :type seen_oids: set
         :return: Templated content
         """
-
         if not jinja_env:
             jinja_env = self.get_template_env()
 
@@ -1033,7 +1030,6 @@ def get_flat_relative_ids(self,
         """
         Get a flat set of relatives' ids, either upstream or downstream.
         """
-
         if not self._dag:
             return set()
 

diff --git a/airflow/models/dag.py b/airflow/models/dag.py
@@ -825,7 +825,6 @@ def resolve_template_files(self):
 
     def get_template_env(self) -> jinja2.Environment:
         """Build a Jinja2 environment."""
-
         # Collect directories to search for template files
         searchpath = [self.folder]
         if self.template_searchpath:
@@ -1682,7 +1681,6 @@ def deactivate_unknown_dags(active_dag_ids, session=None):
         :type active_dag_ids: list[unicode]
         :return: None
         """
-
         if len(active_dag_ids) == 0:
             return
         for dag in session.query(

diff --git a/airflow/models/dagbag.py b/airflow/models/dagbag.py
@@ -360,7 +360,6 @@ def bag_dag(self, dag, root_dag):
         Adds the DAG into the bag, recurses into sub dags.
         Throws AirflowDagCycleException if a cycle is detected in this dag or its subdags
         """
-
         test_cycle(dag)  # throws if a task cycle is found
 
         dag.resolve_template_files()

diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
@@ -277,7 +277,6 @@ def get_dag(self):
     @provide_session
     def get_previous_dagrun(self, state: Optional[str] = None, session: Session = None) -> Optional['DagRun']:
         """The previous DagRun, if there is one"""
-
         filters = [
             DagRun.dag_id == self.dag_id,
             DagRun.execution_date < self.execution_date,
@@ -311,7 +310,6 @@ def update_state(self, session: Session = None) -> List[TI]:
         :return: ready_tis: the tis that can be scheduled in the current loop
         :rtype ready_tis: list[airflow.models.TaskInstance]
         """
-
         dag = self.get_dag()
         ready_tis: List[TI] = []
         tis = list(self.get_task_instances(session=session, state=State.task_states + (State.SHUTDOWN,)))

diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
@@ -937,7 +937,6 @@ def check_and_change_state_before_execution(    # pylint: disable=too-many-argum
         :return: whether the state was changed to running or not
         :rtype: bool
         """
-
         task = self.task
         self.refresh_from_task(task, pool_override=pool)
         self.test_mode = test_mode
@@ -1060,7 +1059,6 @@ def _run_raw_task(
         :param session: SQLAlchemy ORM Session
         :type session: Session
         """
-
         task = self.task
         self.test_mode = test_mode
         self.refresh_from_task(task, pool_override=pool)
@@ -1716,7 +1714,6 @@ def xcom_push(
             task on a future date without it being immediately visible.
         :type execution_date: datetime
         """
-
         if execution_date and execution_date < self.execution_date:
             raise ValueError(
                 'execution_date can not be in the past (current '
@@ -1765,7 +1762,6 @@ def xcom_pull(      # pylint: disable=inconsistent-return-statements
             are returned as well.
         :type include_prior_dates: bool
         """
-
         if dag_id is None:
             dag_id = self.dag_id
 
@@ -1918,7 +1914,6 @@ def construct_task_instance(self, session=None, lock_for_update=False) -> TaskIn
             session is committed.
         :return: the task instance constructed
         """
-
         qry = session.query(TaskInstance).filter(
             TaskInstance.dag_id == self._dag_id,
             TaskInstance.task_id == self._task_id,

diff --git a/airflow/models/variable.py b/airflow/models/variable.py
@@ -155,7 +155,6 @@ def set(
         :param serialize_json: Serialize the value to a JSON string
         :param session: SQL Alchemy Sessions
         """
-
         if serialize_json:
             stored_value = json.dumps(value, indent=2)
         else:

diff --git a/airflow/operators/sql.py b/airflow/operators/sql.py
@@ -465,7 +465,6 @@ def push(self, meta_data):
         Optional: Send data check info and metadata to an external database.
         Default functionality will log metadata.
         """
-
         info = "\n".join([f"""{key}: {item}""" for key, item in meta_data.items()])
         self.log.info("Log from %s:\n%s", self.dag_id, info)
 

diff --git a/airflow/providers/amazon/aws/hooks/cloud_formation.py b/airflow/providers/amazon/aws/hooks/cloud_formation.py
@@ -42,7 +42,6 @@ def get_stack_status(self, stack_name):
         """
         Get stack status from CloudFormation.
         """
-
         self.log.info('Poking for stack %s', stack_name)
 
         try:
@@ -63,7 +62,6 @@ def create_stack(self, stack_name, params):
         :param params: parameters to be passed to CloudFormation.
         :type params: dict
         """
-
         if 'StackName' not in params:
             params['StackName'] = stack_name
         self.get_conn().create_stack(**params)
@@ -77,7 +75,6 @@ def delete_stack(self, stack_name, params=None):
         :param params: parameters to be passed to CloudFormation (optional).
         :type params: dict
         """
-
         params = params or {}
         if 'StackName' not in params:
             params['StackName'] = stack_name

diff --git a/airflow/providers/amazon/aws/hooks/emr.py b/airflow/providers/amazon/aws/hooks/emr.py
@@ -49,7 +49,6 @@ def get_cluster_id_by_name(self, emr_cluster_name: str, cluster_states: List[str
         :type cluster_states: list
         :return: id of the EMR cluster
         """
-
         response = self.get_conn().list_clusters(ClusterStates=cluster_states)
 
         matching_clusters = list(
@@ -73,7 +72,6 @@ def create_job_flow(self, job_flow_overrides: Dict[str, Any]) -> Dict[str, Any]:
         run_job_flow method.
         Overrides for this config may be passed as the job_flow_overrides.
         """
-
         if not self.emr_conn_id:
             raise AirflowException('emr_conn_id must be present to use create_job_flow')
-Original file line number
+Diff line change
@@ Expand Up / @@ -87,7 +87,6 @@ def get_dag_runs( @@
         """
         Get all DAG Runs.
         """
         query = session.query(DagRun)
         #  This endpoint allows specifying ~ as the dag_id to retrieve DAG Runs for all DAGs.
@@ Expand Down @@