Upgrade black to 20.8b1 (#10818)

apache · Sep 9, 2020 · 9549274 · 9549274
1 parent 004e1d8
commit 9549274
Show file tree

Hide file tree

Showing 334 changed files with 4,868 additions and 1,358 deletions.
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -150,7 +150,7 @@ repos:
  hooks:
  - id: check-hooks-apply
  - repo: https://github.com/psf/black
- rev: 19.10b0
+ rev: 20.8b1
  hooks:
  - id: black
  files: api_connexion/.*\.py|.*providers.*\.py

diff --git a/airflow/api_connexion/endpoints/health_endpoint.py b/airflow/api_connexion/endpoints/health_endpoint.py
@@ -40,7 +40,10 @@ def get_health():
 
  payload = {
  "metadatabase": {"status": metadatabase_status},
- "scheduler": {"status": scheduler_status, "latest_scheduler_heartbeat": latest_scheduler_heartbeat,},
+ "scheduler": {
+ "status": scheduler_status,
+ "latest_scheduler_heartbeat": latest_scheduler_heartbeat,
+ },
  }
 
  return health_schema.dump(payload)
diff --git a/airflow/api_connexion/endpoints/variable_endpoint.py b/airflow/api_connexion/endpoints/variable_endpoint.py
@@ -64,7 +64,12 @@ def get_variables(session, limit: Optional[int], offset: Optional[int] = None) -
  if limit:
  query = query.limit(limit)
  variables = query.all()
- return variable_collection_schema.dump({"variables": variables, "total_entries": total_entries,})
+ return variable_collection_schema.dump(
+ {
+ "variables": variables,
+ "total_entries": total_entries,
+ }
+ )
 
 
 @security.requires_authentication

diff --git a/airflow/api_connexion/schemas/error_schema.py b/airflow/api_connexion/schemas/error_schema.py
@@ -33,7 +33,9 @@ class Meta:
  import_error_id = auto_field("id", dump_only=True)
  timestamp = auto_field(format="iso")
  filename = auto_field()
- stack_trace = auto_field("stacktrace",)
+ stack_trace = auto_field(
+ "stacktrace",
+ )
 
 
 class ImportErrorCollection(NamedTuple):

diff --git a/airflow/providers/amazon/aws/example_dags/example_datasync_1.py b/airflow/providers/amazon/aws/example_dags/example_datasync_1.py
@@ -57,7 +57,9 @@
  # [END howto_operator_datasync_1_1]
 
 with models.DAG(
- "example_datasync_1_2", start_date=days_ago(1), schedule_interval=None, # Override to match your needs
+ "example_datasync_1_2",
+ start_date=days_ago(1),
+ schedule_interval=None, # Override to match your needs
 ) as dag:
  # [START howto_operator_datasync_1_2]
  datasync_task_2 = AWSDataSyncOperator(

diff --git a/airflow/providers/amazon/aws/example_dags/example_ecs_fargate.py b/airflow/providers/amazon/aws/example_dags/example_ecs_fargate.py
@@ -56,7 +56,12 @@
  task_definition="hello-world",
  launch_type="FARGATE",
  overrides={
- "containerOverrides": [{"name": "hello-world-container", "command": ["echo", "hello", "world"],},],
+ "containerOverrides": [
+ {
+ "name": "hello-world-container",
+ "command": ["echo", "hello", "world"],
+ },
+ ],
  },
  network_configuration={
  "awsvpcConfiguration": {

diff --git a/airflow/providers/amazon/aws/example_dags/example_s3_bucket.py b/airflow/providers/amazon/aws/example_dags/example_s3_bucket.py
@@ -31,7 +31,9 @@ def upload_keys():
  s3_hook = S3Hook()
  for i in range(0, 3):
  s3_hook.load_string(
- string_data="input", key=f"path/data{i}", bucket_name=BUCKET_NAME,
+ string_data="input",
+ key=f"path/data{i}",
+ bucket_name=BUCKET_NAME,
  )
 
 
@@ -44,15 +46,19 @@ def upload_keys():
 ) as dag:
 
  create_bucket = S3CreateBucketOperator(
- task_id='s3_bucket_dag_create', bucket_name=BUCKET_NAME, region_name='us-east-1',
+ task_id='s3_bucket_dag_create',
+ bucket_name=BUCKET_NAME,
+ region_name='us-east-1',
  )
 
  add_keys_to_bucket = PythonOperator(
  task_id="s3_bucket_dag_add_keys_to_bucket", python_callable=upload_keys
  )
 
  delete_bucket = S3DeleteBucketOperator(
- task_id='s3_bucket_dag_delete', bucket_name=BUCKET_NAME, force_delete=True,
+ task_id='s3_bucket_dag_delete',
+ bucket_name=BUCKET_NAME,
+ force_delete=True,
  )
 
  create_bucket >> add_keys_to_bucket >> delete_bucket
diff --git a/airflow/providers/amazon/aws/hooks/base_aws.py b/airflow/providers/amazon/aws/hooks/base_aws.py
@@ -71,7 +71,9 @@ def _create_basic_session(self, session_kwargs: Dict[str, Any]) -> boto3.session
  self.log.info("Retrieving region_name from Connection.extra_config['region_name']")
  region_name = self.extra_config["region_name"]
  self.log.info(
- "Creating session with aws_access_key_id=%s region_name=%s", aws_access_key_id, region_name,
+ "Creating session with aws_access_key_id=%s region_name=%s",
+ aws_access_key_id,
+ region_name,
  )
 
  return boto3.session.Session(
@@ -161,7 +163,9 @@ def _assume_role(
  assume_role_kwargs["ExternalId"] = self.extra_config.get("external_id")
  role_session_name = f"Airflow_{self.conn.conn_id}"
  self.log.info(
- "Doing sts_client.assume_role to role_arn=%s (role_session_name=%s)", role_arn, role_session_name,
+ "Doing sts_client.assume_role to role_arn=%s (role_session_name=%s)",
+ role_arn,
+ role_session_name,
  )
  return sts_client.assume_role(
  RoleArn=role_arn, RoleSessionName=role_session_name, **assume_role_kwargs
@@ -317,7 +321,8 @@ def _get_credentials(self, region_name):
  # http://boto3.readthedocs.io/en/latest/guide/configuration.html
 
  self.log.info(
- "Creating session using boto3 credential strategy region_name=%s", region_name,
+ "Creating session using boto3 credential strategy region_name=%s",
+ region_name,
  )
  session = boto3.session.Session(region_name=region_name)
  return session, None

diff --git a/airflow/providers/amazon/aws/hooks/batch_client.py b/airflow/providers/amazon/aws/hooks/batch_client.py
@@ -342,7 +342,10 @@ def poll_job_status(self, job_id: str, match_status: List[str]) -> bool:
  job = self.get_job_description(job_id)
  job_status = job.get("status")
  self.log.info(
- "AWS Batch job (%s) check status (%s) in %s", job_id, job_status, match_status,
+ "AWS Batch job (%s) check status (%s) in %s",
+ job_id,
+ job_status,
+ match_status,
  )
 
  if job_status in match_status:

diff --git a/airflow/providers/amazon/aws/hooks/redshift.py b/airflow/providers/amazon/aws/hooks/redshift.py
@@ -116,6 +116,7 @@ def create_cluster_snapshot(self, snapshot_identifier: str, cluster_identifier:
  :type cluster_identifier: str
  """
  response = self.get_conn().create_cluster_snapshot(
- SnapshotIdentifier=snapshot_identifier, ClusterIdentifier=cluster_identifier,
+ SnapshotIdentifier=snapshot_identifier,
+ ClusterIdentifier=cluster_identifier,
  )
  return response['Snapshot'] if response['Snapshot'] else None
diff --git a/airflow/providers/amazon/aws/operators/batch.py b/airflow/providers/amazon/aws/operators/batch.py
@@ -152,7 +152,9 @@ def submit_job(self, context: Dict): # pylint: disable=unused-argument
  :raises: AirflowException
  """
  self.log.info(
- "Running AWS Batch job - job definition: %s - on queue %s", self.job_definition, self.job_queue,
+ "Running AWS Batch job - job definition: %s - on queue %s",
+ self.job_definition,
+ self.job_queue,
  )
  self.log.info("AWS Batch job - container overrides: %s", self.overrides)
 

diff --git a/airflow/providers/amazon/aws/operators/datasync.py b/airflow/providers/amazon/aws/operators/datasync.py
@@ -182,7 +182,8 @@ def get_hook(self):
  """
  if not self.hook:
  self.hook = AWSDataSyncHook(
- aws_conn_id=self.aws_conn_id, wait_interval_seconds=self.wait_interval_seconds,
+ aws_conn_id=self.aws_conn_id,
+ wait_interval_seconds=self.wait_interval_seconds,
  )
  return self.hook
 
@@ -239,7 +240,8 @@ def _get_tasks_and_locations(self):
 
  self.log.info("Finding DataSync TaskArns that have these LocationArns")
  self.candidate_task_arns = hook.get_task_arns_for_location_arns(
- self.candidate_source_location_arns, self.candidate_destination_location_arns,
+ self.candidate_source_location_arns,
+ self.candidate_destination_location_arns,
  )
  self.log.info("Found candidate DataSync TaskArns %s", self.candidate_task_arns)
 

diff --git a/airflow/providers/amazon/aws/operators/ec2_start_instance.py b/airflow/providers/amazon/aws/operators/ec2_start_instance.py
@@ -65,5 +65,7 @@ def execute(self, context):
  instance = ec2_hook.get_instance(instance_id=self.instance_id)
  instance.start()
  ec2_hook.wait_for_state(
- instance_id=self.instance_id, target_state="running", check_interval=self.check_interval,
+ instance_id=self.instance_id,
+ target_state="running",
+ check_interval=self.check_interval,
  )
diff --git a/airflow/providers/amazon/aws/operators/ec2_stop_instance.py b/airflow/providers/amazon/aws/operators/ec2_stop_instance.py
@@ -65,5 +65,7 @@ def execute(self, context):
  instance = ec2_hook.get_instance(instance_id=self.instance_id)
  instance.stop()
  ec2_hook.wait_for_state(
- instance_id=self.instance_id, target_state="stopped", check_interval=self.check_interval,
+ instance_id=self.instance_id,
+ target_state="stopped",
+ check_interval=self.check_interval,
  )
diff --git a/airflow/providers/amazon/aws/secrets/secrets_manager.py b/airflow/providers/amazon/aws/secrets/secrets_manager.py
@@ -85,7 +85,9 @@ def client(self):
  """
  Create a Secrets Manager client
  """
- session = boto3.session.Session(profile_name=self.profile_name,)
+ session = boto3.session.Session(
+ profile_name=self.profile_name,
+ )
  return session.client(service_name="secretsmanager", **self.kwargs)
 
  def get_conn_uri(self, conn_id: str) -> Optional[str]:
@@ -126,7 +128,9 @@ def _get_secret(self, path_prefix: str, secret_id: str) -> Optional[str]:
  """
  secrets_path = self.build_path(path_prefix, secret_id, self.sep)
  try:
- response = self.client.get_secret_value(SecretId=secrets_path,)
+ response = self.client.get_secret_value(
+ SecretId=secrets_path,
+ )
  return response.get('SecretString')
  except self.client.exceptions.ResourceNotFoundException:
  self.log.debug(

diff --git a/airflow/providers/amazon/aws/transfers/dynamodb_to_s3.py b/airflow/providers/amazon/aws/transfers/dynamodb_to_s3.py
@@ -42,7 +42,9 @@ def _upload_file_to_s3(file_obj, bucket_name, s3_key_prefix):
  s3_client = S3Hook().get_conn()
  file_obj.seek(0)
  s3_client.upload_file(
- Filename=file_obj.name, Bucket=bucket_name, Key=s3_key_prefix + str(uuid4()),
+ Filename=file_obj.name,
+ Bucket=bucket_name,
+ Key=s3_key_prefix + str(uuid4()),
  )
 
 

diff --git a/airflow/providers/amazon/aws/transfers/s3_to_sftp.py b/airflow/providers/amazon/aws/transfers/s3_to_sftp.py
@@ -61,8 +61,7 @@ def __init__(
 
  @staticmethod
  def get_s3_key(s3_key):
- """This parses the correct format for S3 keys regardless of how the S3 url is passed.
- """
+ """This parses the correct format for S3 keys regardless of how the S3 url is passed."""
 
  parsed_s3_key = urlparse(s3_key)
  return parsed_s3_key.path.lstrip('/')

diff --git a/airflow/providers/apache/druid/transfers/hive_to_druid.py b/airflow/providers/apache/druid/transfers/hive_to_druid.py
@@ -152,7 +152,10 @@ def execute(self, context: Dict[str, Any]) -> None:
  druid = DruidHook(druid_ingest_conn_id=self.druid_ingest_conn_id)
 
  try:
- index_spec = self.construct_ingest_query(static_path=static_path, columns=columns,)
+ index_spec = self.construct_ingest_query(
+ static_path=static_path,
+ columns=columns,
+ )
 
  self.log.info("Inserting rows into Druid, hdfs path: %s", static_path)
 

diff --git a/airflow/providers/apache/hive/operators/hive_stats.py b/airflow/providers/apache/hive/operators/hive_stats.py
@@ -181,5 +181,13 @@ def execute(self, context: Optional[Dict[str, Any]] = None) -> None:
  mysql.insert_rows(
  table='hive_stats',
  rows=rows,
- target_fields=['ds', 'dttm', 'table_name', 'partition_repr', 'col', 'metric', 'value',],
+ target_fields=[
+ 'ds',
+ 'dttm',
+ 'table_name',
+ 'partition_repr',
+ 'col',
+ 'metric',
+ 'value',
+ ],
  )
diff --git a/airflow/providers/apache/livy/example_dags/example_livy.py b/airflow/providers/apache/livy/example_dags/example_livy.py
@@ -28,7 +28,10 @@
 args = {'owner': 'airflow', 'email': ['[email protected]'], 'depends_on_past': False}
 
 with DAG(
- dag_id='example_livy_operator', default_args=args, schedule_interval='@daily', start_date=days_ago(5),
+ dag_id='example_livy_operator',
+ default_args=args,
+ schedule_interval='@daily',
+ start_date=days_ago(5),
 ) as dag:
 
  livy_java_task = LivyOperator(
@@ -38,7 +41,9 @@
  file='/spark-examples.jar',
  args=[10],
  num_executors=1,
- conf={'spark.shuffle.compress': 'false',},
+ conf={
+ 'spark.shuffle.compress': 'false',
+ },
  class_name='org.apache.spark.examples.SparkPi',
  )
 

diff --git a/airflow/providers/apache/pig/example_dags/example_pig.py b/airflow/providers/apache/pig/example_dags/example_pig.py
@@ -34,4 +34,9 @@
  tags=['example'],
 )
 
-run_this = PigOperator(task_id="run_example_pig_script", pig="ls /;", pig_opts="-x local", dag=dag,)
+run_this = PigOperator(
+ task_id="run_example_pig_script",
+ pig="ls /;",
+ pig_opts="-x local",
+ dag=dag,
+)
diff --git a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
@@ -133,7 +133,10 @@ def get_namespace(self):
  return namespace
 
  def get_pod_log_stream(
- self, pod_name: str, container: Optional[str] = "", namespace: Optional[str] = None,
+ self,
+ pod_name: str,
+ container: Optional[str] = "",
+ namespace: Optional[str] = None,
  ) -> Tuple[watch.Watch, Generator[str, None, None]]:
  """
  Retrieves a log stream for a container in a kubernetes pod.
@@ -159,7 +162,10 @@ def get_pod_log_stream(
  )
 
  def get_pod_logs(
- self, pod_name: str, container: Optional[str] = "", namespace: Optional[str] = None,
+ self,
+ pod_name: str,
+ container: Optional[str] = "",
+ namespace: Optional[str] = None,
  ):
  """
  Retrieves a container's log from the specified pod.

diff --git a/airflow/providers/databricks/example_dags/example_databricks.py b/airflow/providers/databricks/example_dags/example_databricks.py
@@ -57,7 +57,9 @@
 
  notebook_task_params = {
  'new_cluster': new_cluster,
- 'notebook_task': {'notebook_path': '/Users/[email protected]/PrepareData',},
+ 'notebook_task': {
+ 'notebook_path': '/Users/[email protected]/PrepareData',
+ },
  }
  # Example of using the JSON parameter to initialize the operator.
  notebook_task = DatabricksSubmitRunOperator(task_id='notebook_task', json=notebook_task_params)

diff --git a/airflow/providers/databricks/operators/databricks.py b/airflow/providers/databricks/operators/databricks.py
@@ -43,7 +43,13 @@ def _deep_string_coerce(content, json_path='json'):
  coerce = _deep_string_coerce
  if isinstance(content, str):
  return content
- elif isinstance(content, (int, float,)):
+ elif isinstance(
+ content,
+ (
+ int,
+ float,
+ ),
+ ):
  # Databricks can tolerate either numeric or string types in the API backend.
  return str(content)
  elif isinstance(content, (list, tuple)):

diff --git a/airflow/providers/facebook/ads/hooks/ads.py b/airflow/providers/facebook/ads/hooks/ads.py
@@ -59,7 +59,11 @@ class FacebookAdsReportingHook(BaseHook):
 
  """
 
- def __init__(self, facebook_conn_id: str = "facebook_default", api_version: str = "v6.0",) -> None:
+ def __init__(
+ self,
+ facebook_conn_id: str = "facebook_default",
+ api_version: str = "v6.0",
+ ) -> None:
  super().__init__()
  self.facebook_conn_id = facebook_conn_id
  self.api_version = api_version
@@ -92,7 +96,10 @@ def facebook_ads_config(self) -> Dict:
  return config
 
  def bulk_facebook_report(
- self, params: Dict[str, Any], fields: List[str], sleep_time: int = 5,
+ self,
+ params: Dict[str, Any],
+ fields: List[str],
+ sleep_time: int = 5,
  ) -> List[AdsInsights]:
  """
  Pulls data from the Facebook Ads API

diff --git a/airflow/providers/google/ads/transfers/ads_to_gcs.py b/airflow/providers/google/ads/transfers/ads_to_gcs.py
@@ -124,6 +124,9 @@ def execute(self, context: Dict):
 
  hook = GCSHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
  hook.upload(
- bucket_name=self.bucket, object_name=self.obj, filename=csvfile.name, gzip=self.gzip,
+ bucket_name=self.bucket,
+ object_name=self.obj,
+ filename=csvfile.name,
+ gzip=self.gzip,
  )
  self.log.info("%s uploaded to GCS", self.obj)