Enable specifying dictionary paths in template_fields_renderers

BREEZE.rst

@@ -1830,10 +1830,15 @@ This is the current syntax for  `./breeze <./breeze>`_:
                  9.6 10 11 12 13
 
   --mysql-version MYSQL_VERSION
-          Mysql version used. One of:
+          MySql version used. One of:
 
                  5.7 8
 
+  --mssql-version MSSQL_VERSION
+          MSSql version used. One of:
+
+                 2017-latest 2019-latest
+
   -v, --verbose
           Show verbose information about executed docker, kind, kubectl, helm commands. Useful for
           debugging - when you run breeze with --verbose flags you will be able to see the commands
@@ -2299,10 +2304,15 @@ This is the current syntax for  `./breeze <./breeze>`_:
                  9.6 10 11 12 13
 
   --mysql-version MYSQL_VERSION
-          Mysql version used. One of:
+          MySql version used. One of:
 
                  5.7 8
 
+  --mssql-version MSSQL_VERSION
+          MSSql version used. One of:
+
+                 2017-latest 2019-latest
+
   ****************************************************************************************************
    Enable production image
 

airflow/executors/celery_executor.py

@@ -46,7 +46,7 @@
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, AirflowTaskTimeout
 from airflow.executors.base_executor import BaseExecutor, CommandType, EventBufferValueType
-from airflow.models.taskinstance import SimpleTaskInstance, TaskInstance, TaskInstanceKey
+from airflow.models.taskinstance import TaskInstance, TaskInstanceKey
 from airflow.stats import Stats
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.net import get_hostname
@@ -154,15 +154,15 @@ def __init__(self, exception: Exception, exception_traceback: str):
 
 
 # Task instance that is sent over Celery queues
-# TaskInstanceKey, SimpleTaskInstance, Command, queue_name, CallableTask
-TaskInstanceInCelery = Tuple[TaskInstanceKey, SimpleTaskInstance, CommandType, Optional[str], Task]
+# TaskInstanceKey, Command, queue_name, CallableTask
+TaskInstanceInCelery = Tuple[TaskInstanceKey, CommandType, Optional[str], Task]
 
 
 def send_task_to_executor(
     task_tuple: TaskInstanceInCelery,
 ) -> Tuple[TaskInstanceKey, CommandType, Union[AsyncResult, ExceptionWithTraceback]]:
     """Sends task to executor."""
-    key, _, command, queue, task_to_run = task_tuple
+    key, command, queue, task_to_run = task_tuple
     try:
         with timeout(seconds=OPERATION_TIMEOUT):
             result = task_to_run.apply_async(args=[command], queue=queue)
@@ -250,8 +250,8 @@ def trigger_tasks(self, open_slots: int) -> None:
         task_tuples_to_send: List[TaskInstanceInCelery] = []
 
         for _ in range(min(open_slots, len(self.queued_tasks))):
-            key, (command, _, queue, simple_ti) = sorted_queue.pop(0)
-            task_tuple = (key, simple_ti, command, queue, execute_command)
+            key, (command, _, queue, _) = sorted_queue.pop(0)
+            task_tuple = (key, command, queue, execute_command)
             task_tuples_to_send.append(task_tuple)
             if key not in self.task_publish_retries:
                 self.task_publish_retries[key] = 1
@@ -260,7 +260,7 @@ def trigger_tasks(self, open_slots: int) -> None:
             self._process_tasks(task_tuples_to_send)
 
     def _process_tasks(self, task_tuples_to_send: List[TaskInstanceInCelery]) -> None:
-        first_task = next(t[4] for t in task_tuples_to_send)
+        first_task = next(t[3] for t in task_tuples_to_send)
 
         # Celery state queries will stuck if we do not use one same backend
         # for all tasks.

airflow/providers/airbyte/CHANGELOG.rst

@@ -19,6 +19,29 @@
 Changelog
 ---------
 
+2.1.0
+.....
+
+Breaking changes
+~~~~~~~~~~~~~~~~
+
+
+Features
+~~~~~~~~
+
+
+Bug Fixes
+~~~~~~~~~
+
+* ``Fixed wrongly escaped characters in amazon's changelog (#17020)``
+
+.. Below changes are excluded from the changelog. Move them to
+   appropriate section above if needed. Do not delete the lines(!):
+   * ``AirbyteHook - Consider incomplete status (#16965)``
+   * ``Prepare documentation for July release of providers. (#17015)``
+   * ``Updating Airbyte example DAG to use XComArgs (#16867)``
+   * ``Removes pylint from our toolchain (#16682)``
+
 2.0.0
 .....
 

airflow/providers/airbyte/provider.yaml

@@ -22,6 +22,7 @@ description: |
   `Airbyte <https://airbyte.io/>`__
 
 versions:
+  - 2.1.0
   - 2.0.0
   - 1.0.0
 

airflow/providers/amazon/CHANGELOG.rst

@@ -29,19 +29,28 @@ Features
 * ``Added select_query to the templated fields in RedshiftToS3Operator (#16767)``
 * ``AWS Hook - allow IDP HTTP retry (#12639) (#16612)``
 * ``Update Boto3 API calls in ECSOperator (#16050)``
+* ``Adding custom Salesforce connection type + SalesforceToS3Operator updates (#17162)``
+* ``Adding SalesforceToS3Operator to Amazon Provider (#17094)``
 
 Bug Fixes
 ~~~~~~~~~
 
 * ``AWS DataSync default polling adjusted from 5s to 30s (#11011)``
 * ``Fix wrong template_fields_renderers for AWS operators (#16820)``
 * ``AWS DataSync cancel task on exception (#11011) (#16589)``
+* ``Fixed template_fields_renderers for Amazon provider (#17087)``
+* ``removing try-catch block (#17081)``
+* ``ECSOperator / pass context to self.xcom_pull as it was missing (when using reattach) (#17141)``
+* ``Made S3ToRedshiftOperator transaction safe (#17117)``
 
 .. Below changes are excluded from the changelog. Move them to
    appropriate section above if needed. Do not delete the lines(!):
    * ``Removes pylint from our toolchain (#16682)``
    * ``Bump sphinxcontrib-spelling and minor improvements (#16675)``
    * ``Prepare documentation for July release of providers. (#17015)``
+   * ``Added docs & doc ref's for AWS transfer operators between SFTP & S3 (#16964)``
+   * ``Fixed wrongly escaped characters in amazon's changelog (#17020)``
+   * ``Updating Amazon-AWS example DAGs to use XComArgs (#16868)``
 
 2.0.0
 .....

airflow/providers/amazon/aws/example_dags/example_dms_full_load_task.py

@@ -47,17 +47,16 @@
     ]
 }
 
-DEFAULT_ARGS = {
-    'owner': 'airflow',
-    'depends_on_past': False,
-    'email': ['airflow@example.com'],
-    'email_on_failure': False,
-    'email_on_retry': False,
-}
 
 with DAG(
     dag_id='dms_full_load_task_run_dag',
-    default_args=DEFAULT_ARGS,
+    default_args={
+        'owner': 'airflow',
+        'depends_on_past': False,
+        'email': ['airflow@example.com'],
+        'email_on_failure': False,
+        'email_on_retry': False,
+    },
     dagrun_timeout=timedelta(hours=2),
     start_date=days_ago(2),
     schedule_interval='0 3 * * *',
@@ -78,22 +77,27 @@
     # [START howto_dms_start_task_operator]
     start_task = DmsStartTaskOperator(
         task_id='start_task',
-        replication_task_arn='{{ task_instance.xcom_pull(task_ids="create_task", key="return_value") }}',
+        replication_task_arn=create_task.output,
     )
     # [END howto_dms_start_task_operator]
 
     # [START howto_dms_task_completed_sensor]
     wait_for_completion = DmsTaskCompletedSensor(
         task_id='wait_for_completion',
-        replication_task_arn='{{ task_instance.xcom_pull(task_ids="create_task", key="return_value") }}',
+        replication_task_arn=create_task.output,
     )
     # [END howto_dms_task_completed_sensor]
 
     # [START howto_dms_delete_task_operator]
     delete_task = DmsDeleteTaskOperator(
         task_id='delete_task',
-        replication_task_arn='{{ task_instance.xcom_pull(task_ids="create_task", key="return_value") }}',
+        replication_task_arn=create_task.output,
     )
     # [END howto_dms_delete_task_operator]
 
-    create_task >> start_task >> wait_for_completion >> delete_task
+    start_task >> wait_for_completion >> delete_task
+
+    # Task dependencies created via `XComArgs`:
+    #   create_task >> start_task
+    #   create_task >> wait_for_completion
+    #   create_task >> delete_task

airflow/providers/amazon/aws/example_dags/example_ecs_fargate.py

@@ -28,17 +28,15 @@
 from airflow import DAG
 from airflow.providers.amazon.aws.operators.ecs import ECSOperator
 
-DEFAULT_ARGS = {
-    "owner": "airflow",
-    "depends_on_past": False,
-    "email": ["airflow@example.com"],
-    "email_on_failure": False,
-    "email_on_retry": False,
-}
-
 dag = DAG(
     dag_id="ecs_fargate_dag",
-    default_args=DEFAULT_ARGS,
+    default_args={
+        "owner": "airflow",
+        "depends_on_past": False,
+        "email": ["airflow@example.com"],
+        "email_on_failure": False,
+        "email_on_retry": False,
+    },
     default_view="graph",
     schedule_interval=None,
     start_date=datetime.datetime(2020, 1, 1),

airflow/providers/amazon/aws/example_dags/example_emr_job_flow_automatic_steps.py

@@ -25,14 +25,6 @@
 from airflow.providers.amazon.aws.sensors.emr_job_flow import EmrJobFlowSensor
 from airflow.utils.dates import days_ago
 
-DEFAULT_ARGS = {
-    'owner': 'airflow',
-    'depends_on_past': False,
-    'email': ['airflow@example.com'],
-    'email_on_failure': False,
-    'email_on_retry': False,
-}
-
 # [START howto_operator_emr_automatic_steps_config]
 SPARK_STEPS = [
     {
@@ -69,7 +61,13 @@
 
 with DAG(
     dag_id='emr_job_flow_automatic_steps_dag',
-    default_args=DEFAULT_ARGS,
+    default_args={
+        'owner': 'airflow',
+        'depends_on_past': False,
+        'email': ['airflow@example.com'],
+        'email_on_failure': False,
+        'email_on_retry': False,
+    },
     dagrun_timeout=timedelta(hours=2),
     start_date=days_ago(2),
     schedule_interval='0 3 * * *',
@@ -86,9 +84,10 @@
 
     job_sensor = EmrJobFlowSensor(
         task_id='check_job_flow',
-        job_flow_id="{{ task_instance.xcom_pull(task_ids='create_job_flow', key='return_value') }}",
+        job_flow_id=job_flow_creator.output,
         aws_conn_id='aws_default',
     )
-
-    job_flow_creator >> job_sensor
     # [END howto_operator_emr_automatic_steps_tasks]
+
+    # Task dependency created via `XComArgs`:
+    #   job_flow_creator >> job_sensor

airflow/providers/amazon/aws/example_dags/example_emr_job_flow_manual_steps.py

@@ -30,14 +30,6 @@
 from airflow.providers.amazon.aws.sensors.emr_step import EmrStepSensor
 from airflow.utils.dates import days_ago
 
-DEFAULT_ARGS = {
-    'owner': 'airflow',
-    'depends_on_past': False,
-    'email': ['airflow@example.com'],
-    'email_on_failure': False,
-    'email_on_retry': False,
-}
-
 SPARK_STEPS = [
     {
         'Name': 'calculate_pi',
@@ -71,7 +63,13 @@
 
 with DAG(
     dag_id='emr_job_flow_manual_steps_dag',
-    default_args=DEFAULT_ARGS,
+    default_args={
+        'owner': 'airflow',
+        'depends_on_past': False,
+        'email': ['airflow@example.com'],
+        'email_on_failure': False,
+        'email_on_retry': False,
+    },
     dagrun_timeout=timedelta(hours=2),
     start_date=days_ago(2),
     schedule_interval='0 3 * * *',
@@ -88,23 +86,28 @@
 
     step_adder = EmrAddStepsOperator(
         task_id='add_steps',
-        job_flow_id="{{ task_instance.xcom_pull(task_ids='create_job_flow', key='return_value') }}",
+        job_flow_id=cluster_creator.output,
         aws_conn_id='aws_default',
         steps=SPARK_STEPS,
     )
 
     step_checker = EmrStepSensor(
         task_id='watch_step',
-        job_flow_id="{{ task_instance.xcom_pull('create_job_flow', key='return_value') }}",
+        job_flow_id=cluster_creator.output,
         step_id="{{ task_instance.xcom_pull(task_ids='add_steps', key='return_value')[0] }}",
         aws_conn_id='aws_default',
     )
 
     cluster_remover = EmrTerminateJobFlowOperator(
         task_id='remove_cluster',
-        job_flow_id="{{ task_instance.xcom_pull(task_ids='create_job_flow', key='return_value') }}",
+        job_flow_id=cluster_creator.output,
         aws_conn_id='aws_default',
     )
 
-    cluster_creator >> step_adder >> step_checker >> cluster_remover
+    step_adder >> step_checker >> cluster_remover
     # [END howto_operator_emr_manual_steps_tasks]
+
+    # Task dependencies created via `XComArgs`:
+    #   cluster_creator >> step_adder
+    #   cluster_creator >> step_checker
+    #   cluster_creator >> cluster_remover

airflow/providers/amazon/aws/example_dags/example_google_api_to_s3_transfer_advanced.py

@@ -37,7 +37,7 @@
 
 from airflow import DAG
 from airflow.operators.dummy import DummyOperator
-from airflow.operators.python import BranchPythonOperator
+from airflow.operators.python import BranchPythonOperator, get_current_context
 from airflow.providers.amazon.aws.transfers.google_api_to_s3 import GoogleApiToS3Operator
 from airflow.utils.dates import days_ago
 
@@ -53,12 +53,13 @@
 
 
 # [START howto_operator_google_api_to_s3_transfer_advanced_task_1_2]
-def _check_and_transform_video_ids(xcom_key, task_ids, task_instance, **kwargs):
-    video_ids_response = task_instance.xcom_pull(task_ids=task_ids, key=xcom_key)
+def _check_and_transform_video_ids(task_output):
+    video_ids_response = task_output
     video_ids = [item['id']['videoId'] for item in video_ids_response['items']]
 
     if video_ids:
-        task_instance.xcom_push(key='video_ids', value={'id': ','.join(video_ids)})
+        context = get_current_context()
+        context["task_instance"].xcom_push(key='video_ids', value={'id': ','.join(video_ids)})
         return 'video_data_to_s3'
     return 'no_video_ids'
 
@@ -98,7 +99,7 @@ def _check_and_transform_video_ids(xcom_key, task_ids, task_instance, **kwargs):
     # [START howto_operator_google_api_to_s3_transfer_advanced_task_1_1]
     task_check_and_transform_video_ids = BranchPythonOperator(
         python_callable=_check_and_transform_video_ids,
-        op_args=[task_video_ids_to_s3.google_api_response_via_xcom, task_video_ids_to_s3.task_id],
+        op_args=[task_video_ids_to_s3.output[task_video_ids_to_s3.google_api_response_via_xcom]],
         task_id='check_and_transform_video_ids',
     )
     # [END howto_operator_google_api_to_s3_transfer_advanced_task_1_1]
@@ -121,4 +122,7 @@ def _check_and_transform_video_ids(xcom_key, task_ids, task_instance, **kwargs):
     # [START howto_operator_google_api_to_s3_transfer_advanced_task_2_1]
     task_no_video_ids = DummyOperator(task_id='no_video_ids')
     # [END howto_operator_google_api_to_s3_transfer_advanced_task_2_1]
-    task_video_ids_to_s3 >> task_check_and_transform_video_ids >> [task_video_data_to_s3, task_no_video_ids]
+    task_check_and_transform_video_ids >> [task_video_data_to_s3, task_no_video_ids]
+
+    # Task dependency created via `XComArgs`:
+    #   task_video_ids_to_s3 >> task_check_and_transform_video_ids

airflow/providers/apache/beam/.latest-doc-only-change.txt

@@ -1 +1 @@
-b916b7507921129dc48d6add1bdc4b923b60c9b9
+d02ded65eaa7d2281e249b3fa028605d1b4c52fb

airflow/providers/apache/cassandra/.latest-doc-only-change.txt

@@ -1 +1 @@
-b916b7507921129dc48d6add1bdc4b923b60c9b9
+91f4d80ff09093de49478214c5bd027e02c92a0e