More improvements in the Databricks operators

airflow/providers/databricks/operators/databricks.py

@@ -526,8 +526,8 @@ class DatabricksRunNowOperator(BaseOperator):
 
         .. seealso::
             https://docs.databricks.com/dev-tools/api/latest/jobs.html#operation/JobsRunNow
-    :param python_named_parameters: A list of parameters for jobs with python wheel tasks,
-        e.g. "python_named_parameters": {"name": "john doe", "age":  "35"}.
+    :param python_named_params: A list of named parameters for jobs with python wheel tasks,
+        e.g. "python_named_params": {"name": "john doe", "age":  "35"}.
         If specified upon run-now, it would overwrite the parameters specified in job setting.
         This field will be templated.
 
@@ -562,7 +562,7 @@ class DatabricksRunNowOperator(BaseOperator):
         token based authentication, provide the key ``token`` in the extra field for the
         connection and create the key ``host`` and leave the ``host`` field empty. (templated)
     :param polling_period_seconds: Controls the rate which we poll for the result of
-        this run. By default the operator will poll every 30 seconds.
+        this run. By default, the operator will poll every 30 seconds.
     :param databricks_retry_limit: Amount of times retry if the Databricks backend is
         unreachable. Its value must be greater than or equal to 1.
     :param databricks_retry_delay: Number of seconds to wait between retries (it
@@ -590,7 +590,7 @@ def __init__(
         python_params: Optional[List[str]] = None,
         jar_params: Optional[List[str]] = None,
         spark_submit_params: Optional[List[str]] = None,
-        python_named_parameters: Optional[Dict[str, str]] = None,
+        python_named_params: Optional[Dict[str, str]] = None,
         idempotency_token: Optional[str] = None,
         databricks_conn_id: str = 'databricks_default',
         polling_period_seconds: int = 30,
@@ -621,8 +621,8 @@ def __init__(
             self.json['notebook_params'] = notebook_params
         if python_params is not None:
             self.json['python_params'] = python_params
-        if python_named_parameters is not None:
-            self.json['python_named_parameters'] = python_named_parameters
+        if python_named_params is not None:
+            self.json['python_named_params'] = python_named_params
         if jar_params is not None:
             self.json['jar_params'] = jar_params
         if spark_submit_params is not None:

airflow/providers/databricks/operators/databricks_repos.py

@@ -46,15 +46,15 @@ class DatabricksReposCreateOperator(BaseOperator):
     :param databricks_conn_id: Reference to the :ref:`Databricks connection <howto/connection:databricks>`.
         By default and in the common case this will be ``databricks_default``. To use
         token based authentication, provide the key ``token`` in the extra field for the
-        connection and create the key ``host`` and leave the ``host`` field empty.
+        connection and create the key ``host`` and leave the ``host`` field empty. (templated)
     :param databricks_retry_limit: Amount of times retry if the Databricks backend is
         unreachable. Its value must be greater than or equal to 1.
     :param databricks_retry_delay: Number of seconds to wait between retries (it
             might be a floating point number).
     """
 
     # Used in airflow.models.BaseOperator
-    template_fields: Sequence[str] = ('repo_path', 'tag', 'branch')
+    template_fields: Sequence[str] = ('repo_path', 'tag', 'branch', 'databricks_conn_id')
 
     __git_providers__ = {
         "github.com": "gitHub",
@@ -173,15 +173,15 @@ class DatabricksReposUpdateOperator(BaseOperator):
     :param databricks_conn_id: Reference to the :ref:`Databricks connection <howto/connection:databricks>`.
         By default and in the common case this will be ``databricks_default``. To use
         token based authentication, provide the key ``token`` in the extra field for the
-        connection and create the key ``host`` and leave the ``host`` field empty.
+        connection and create the key ``host`` and leave the ``host`` field empty.  (templated)
     :param databricks_retry_limit: Amount of times retry if the Databricks backend is
         unreachable. Its value must be greater than or equal to 1.
     :param databricks_retry_delay: Number of seconds to wait between retries (it
             might be a floating point number).
     """
 
     # Used in airflow.models.BaseOperator
-    template_fields: Sequence[str] = ('repo_path', 'tag', 'branch')
+    template_fields: Sequence[str] = ('repo_path', 'tag', 'branch', 'databricks_conn_id')
 
     def __init__(
         self,
@@ -246,15 +246,15 @@ class DatabricksReposDeleteOperator(BaseOperator):
     :param databricks_conn_id: Reference to the :ref:`Databricks connection <howto/connection:databricks>`.
         By default and in the common case this will be ``databricks_default``. To use
         token based authentication, provide the key ``token`` in the extra field for the
-        connection and create the key ``host`` and leave the ``host`` field empty.
+        connection and create the key ``host`` and leave the ``host`` field empty. (templated)
     :param databricks_retry_limit: Amount of times retry if the Databricks backend is
         unreachable. Its value must be greater than or equal to 1.
     :param databricks_retry_delay: Number of seconds to wait between retries (it
             might be a floating point number).
     """
 
     # Used in airflow.models.BaseOperator
-    template_fields: Sequence[str] = ('repo_path',)
+    template_fields: Sequence[str] = ('repo_path', 'databricks_conn_id')
 
     def __init__(
         self,

airflow/providers/databricks/operators/databricks_sql.py

@@ -42,7 +42,7 @@ class DatabricksSqlOperator(BaseOperator):
         :ref:`howto/operator:DatabricksSqlOperator`
 
     :param databricks_conn_id: Reference to
-        :ref:`Databricks connection id<howto/connection:databricks>`
+        :ref:`Databricks connection id<howto/connection:databricks>` (templated)
     :param http_path: Optional string specifying HTTP path of Databricks SQL Endpoint or cluster.
         If not specified, it should be either specified in the Databricks connection's extra parameters,
         or ``sql_endpoint_name`` must be specified.
@@ -65,7 +65,14 @@ class DatabricksSqlOperator(BaseOperator):
     :param csv_params: parameters that will be passed to the ``csv.DictWriter`` class used to write CSV data.
     """
 
-    template_fields: Sequence[str] = ('sql', '_output_path', 'schema', 'catalog', 'http_headers')
+    template_fields: Sequence[str] = (
+        'sql',
+        '_output_path',
+        'schema',
+        'catalog',
+        'http_headers',
+        'databricks_conn_id',
+    )
     template_ext: Sequence[str] = ('.sql',)
     template_fields_renderers = {'sql': 'sql'}
 
@@ -178,7 +185,7 @@ class DatabricksCopyIntoOperator(BaseOperator):
     :param file_format: Required file format. Supported formats are
         ``CSV``, ``JSON``, ``AVRO``, ``ORC``, ``PARQUET``, ``TEXT``, ``BINARYFILE``.
     :param databricks_conn_id: Reference to
-        :ref:`Databricks connection id<howto/connection:databricks>`
+        :ref:`Databricks connection id<howto/connection:databricks>` (templated)
     :param http_path: Optional string specifying HTTP path of Databricks SQL Endpoint or cluster.
         If not specified, it should be either specified in the Databricks connection's extra parameters,
         or ``sql_endpoint_name`` must be specified.
@@ -209,6 +216,7 @@ class DatabricksCopyIntoOperator(BaseOperator):
         '_file_location',
         '_files',
         '_table_name',
+        'databricks_conn_id',
     )
 
     def __init__(