Repository: apache/airflow
Pull request: Add means to Duplicate connections from UI
closes: #12401

---
**Description**

* This change allows users to duplicate connections via UI.
* user can also create duplicates of multiple connections.
* A descriptive flash message will be shown regarding the success or error in any of the connection duplication.




Problem statement:
Duplicate connections UI
In the UI it would be nice to duplicate the selected connection, in the menu "With Selected". 

The copy would be set with the same name plus some suffix (_copy, _1, whatever) and it is useful for when you have a connection with all equal fields except something in a particular one.

Hints from issue discussion:
Sound good to me. Can I assign you to this ticket?
> Sound good to me. Can I assign you to this ticket?

I have no clue how to approach this and right now I don't have time to investigate, sorry
I would like to work on this task, Should I add a new permission or use the can_create permission?
`airflow/www/templates/airflow/model_list.html:L71`
```python
      {% if can_show or can_edit or can_delete %}
        <td><center>
        {{ lib.btn_crud(can_show, can_edit, can_delete, pk, modelview_name, filters) }}
        </center></td>
      {% endif %}
```
I assume that we gonna have a button on the list view and show view we gonna use the create form prefilled with the item that selected.
![image](https://user-images.githubusercontent.com/904829/99997022-dc6a7d00-2dbc-11eb-818b-61a438f8fd33.png)
![image](https://user-images.githubusercontent.com/904829/99997210-20f61880-2dbd-11eb-9ad6-0ef1de58b6f2.png)

I think, you can use can_create perm. 
I am working on this.
@mik-laj could you please assign this to me.

Commit message: 12401 - test added

Patch to review:
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 21d4a820489e1..c35103ce948b4 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -21,6 +21,7 @@
 import json
 import logging
 import math
+import re
 import socket
 import sys
 import traceback
@@ -77,6 +78,7 @@
 from pygments import highlight, lexers
 from pygments.formatters import HtmlFormatter  # noqa pylint: disable=no-name-in-module
 from sqlalchemy import and_, desc, func, or_, union_all
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import joinedload
 from wtforms import SelectField, validators
 from wtforms.validators import InputRequired
@@ -152,7 +154,6 @@ def get_safe_url(url):
         return url_for('Airflow.index')
 
     query = parse_qsl(parsed.query, keep_blank_values=True)
-
     url = parsed._replace(query=urlencode(query)).geturl()
 
     if parsed.scheme in valid_schemes and parsed.netloc in valid_netlocs:
@@ -911,7 +912,6 @@ def dag_details(self, session=None):
         """Get Dag details."""
         dag_id = request.args.get('dag_id')
         dag = current_app.dag_bag.get_dag(dag_id)
-
         title = "DAG Details"
         root = request.args.get('root', '')
 
@@ -2100,6 +2100,7 @@ def graph(self, session=None):
         root = request.args.get('root')
         if root:
             dag = dag.sub_dag(task_ids_or_regex=root, include_upstream=True, include_downstream=False)
+
         arrange = request.args.get('arrange', dag.orientation)
 
         nodes = task_group_to_dict(dag.task_group)
@@ -2203,6 +2204,7 @@ def duration(self, session=None):
         root = request.args.get('root')
         if root:
             dag = dag.sub_dag(task_ids_or_regex=root, include_upstream=True, include_downstream=False)
+
         chart_height = wwwutils.get_chart_height(dag)
         chart = nvd3.lineChart(name="lineChart", x_is_date=True, height=chart_height, width="1200")
         cum_chart = nvd3.lineChart(name="cumLineChart", x_is_date=True, height=chart_height, width="1200")
@@ -2959,6 +2961,7 @@ class ConnectionModelView(AirflowModelView):
         'edit': 'edit',
         'delete': 'delete',
         'action_muldelete': 'delete',
+        'action_mulduplicate': 'create',
     }
 
     base_permissions = [
@@ -3012,6 +3015,57 @@ def action_muldelete(self, items):
         self.update_redirect()
         return redirect(self.get_redirect())
 
+    @action(
+        'mulduplicate',
+        'Duplicate',
+        'Are you sure you want to duplicate the selected connections?',
+        single=False,
+    )
+    @provide_session
+    @auth.has_access(
+        [
+            (permissions.ACTION_CAN_CREATE, permissions.RESOURCE_CONNECTION),
+        ]
+    )
+    def action_mulduplicate(self, connections, session=None):
+        """Duplicate Multiple connections"""
+        for selected_conn in connections:
+            new_conn_id = selected_conn.conn_id
+            matches = re.findall(r"_Copy\((\d+)\)", selected_conn.conn_id)
+            if matches:
+                # replacing the last _Copy(Num) with _Copy(Num+1)
+                new_conn_id = selected_conn.conn_id.replace(
+                    f'_Copy({matches[-1]})', f'_Copy({int(matches[-1]) + 1})'
+                )
+            else:
+                new_conn_id += '_Copy(1)'
+
+            dup_conn = Connection(
+                new_conn_id,
+                selected_conn.conn_type,
+                selected_conn.description,
+                selected_conn.host,
+                selected_conn.login,
+                selected_conn.password,
+                selected_conn.schema,
+                selected_conn.port,
+                selected_conn.extra,
+            )
+
+            try:
+                session.add(dup_conn)
+                session.commit()
+                flash(f"Connection {new_conn_id} added successfully.", "success")
+            except IntegrityError:
+                flash(
+                    f"Connection {new_conn_id} can't be added. Integrity error, probably unique constraint.",
+                    "warning",
+                )
+                session.rollback()
+
+        self.update_redirect()
+        return redirect(self.get_redirect())
+
     def process_form(self, form, is_created):
         """Process form data."""
         conn_type = form.data['conn_type']
diff --git a/tests/www/test_views.py b/tests/www/test_views.py
index febbb01680a0f..4b167bd8c0db7 100644
--- a/tests/www/test_views.py
+++ b/tests/www/test_views.py
@@ -270,6 +270,73 @@ def test_prefill_form_null_extra(self):
         cmv.prefill_form(form=mock_form, pk=1)
 
 
+class TestMultiDuplicateConnectionModelView(TestBase):
+    def setUp(self):
+        super().setUp()
+        conn1 = Connection(
+            conn_id='google_cloud_default',
+            conn_type='Google Cloud',
+            description='Default Google Cloud Connection',
+        )
+
+        # Already copied and modified
+        conn2 = Connection(
+            conn_id='mongodb',
+            conn_type='FTP',
+            description='MongoDB2',
+            host='localhost',
+            schema='airflow',
+            port='5567',
+        )
+
+        conn3 = Connection(
+            conn_id='mysql_connection',
+            conn_type='FTP',
+            description='MongoDB2',
+            host='localhost',
+            schema='airflow',
+            port='3306',
+        )
+
+        conn4 = Connection(
+            conn_id='postgres_connection_Copy(1)',
+            conn_type='FTP',
+            description='Postgres',
+            host='localhost',
+            schema='airflow',
+            port='3306',
+        )
+
+        self.clear_table(Connection)
+        self.session.add_all([conn1, conn2, conn3, conn4])
+        self.session.commit()
+
+    def tearDown(self):
+        self.clear_table(Connection)
+        super().tearDown()
+
+    def test_duplicate_connection(self):
+        """ Test Duplicate multiple connection with suffix"""
+
+        mock_form = mock.Mock()
+        mock_form.data = {"action": "mulduplicate", "rowid": [1, 2, 4]}
+        resp = self.client.post('/connection/action_post', data=mock_form.data, follow_redirects=True)
+
+        expected_result = {
+            'google_cloud_default',
+            'google_cloud_default_Copy(1)',
+            'mongodb',
+            'mongodb_Copy(1)',
+            'mysql_connection',
+            'postgres_connection_Copy(1)',
+            'postgres_connection_Copy(2)',
+        }
+        response = {conn[0].strip() for conn in self.session.query(Connection.conn_id).all()}
+
+        assert resp.status_code == 200
+        assert expected_result == response
+
+
 class TestVariableModelView(TestBase):
     def setUp(self):
         super().setUp()
@@ -285,7 +352,6 @@ def tearDown(self):
         super().tearDown()
 
     def test_can_handle_error_on_decrypt(self):
-
         # create valid variable
         self.client.post('/variable/add', data=self.variable, follow_redirects=True)
 
@@ -293,8 +359,8 @@ def test_can_handle_error_on_decrypt(self):
         Var = models.Variable  # pylint: disable=invalid-name
         (
             self.session.query(Var)
-            .filter(Var.key == self.variable['key'])
-            .update({'val': 'failed_value_not_encrypted'}, synchronize_session=False)
+                .filter(Var.key == self.variable['key'])
+                .update({'val': 'failed_value_not_encrypted'}, synchronize_session=False)
         )
         self.session.commit()
 
@@ -365,7 +431,6 @@ def test_should_list_plugins_on_page_with_details(self):
         self.check_content_in_response("<em>$PLUGINS_FOLDER/</em>test_plugin.py", resp)
 
     def test_should_list_entrypoint_plugins_on_page_with_details(self):
-
         mock_plugin = AirflowPlugin()
         mock_plugin.name = "test_plugin"
         mock_plugin.source = EntryPointSource(
@@ -561,15 +626,15 @@ def test_health(self):
             BaseJob.job_type == 'SchedulerJob',
             BaseJob.state == 'running',
             BaseJob.latest_heartbeat == last_scheduler_heartbeat_for_testing_1,
-        ).delete()
+            ).delete()
         self.session.commit()
 
         # case-2: unhealthy scheduler status - scenario 1 (SchedulerJob is running too slowly)
         last_scheduler_heartbeat_for_testing_2 = timezone.utcnow() - timedelta(minutes=1)
         (
             self.session.query(BaseJob)
-            .filter(BaseJob.job_type == 'SchedulerJob')
-            .update({'latest_heartbeat': last_scheduler_heartbeat_for_testing_2 - timedelta(seconds=1)})
+                .filter(BaseJob.job_type == 'SchedulerJob')
+                .update({'latest_heartbeat': last_scheduler_heartbeat_for_testing_2 - timedelta(seconds=1)})
         )
         self.session.add(
             BaseJob(
@@ -593,7 +658,7 @@ def test_health(self):
             BaseJob.job_type == 'SchedulerJob',
             BaseJob.state == 'running',
             BaseJob.latest_heartbeat == last_scheduler_heartbeat_for_testing_2,
-        ).delete()
+            ).delete()
         self.session.commit()
 
         # case-3: unhealthy scheduler status - scenario 2 (no running SchedulerJob)
@@ -1858,7 +1923,7 @@ def test_with_execution_date_parameter_only(self):
             self.endpoint + f'&execution_date={self.runs[1].execution_date.isoformat()}',
             data=dict(username='test', password='test'),
             follow_redirects=True,
-        )
+            )
         self.test.assertEqual(response.status_code, 200)
         data = response.data.decode('utf-8')
         self.assert_base_date_and_num_runs(
@@ -1881,7 +1946,7 @@ def test_with_base_date_and_num_runs_parameters_only(self):
             self.endpoint + f'&base_date={self.runs[1].execution_date.isoformat()}&num_runs=2',
             data=dict(username='test', password='test'),
             follow_redirects=True,
-        )
+            )
         self.test.assertEqual(response.status_code, 200)
         data = response.data.decode('utf-8')
         self.assert_base_date_and_num_runs(self.runs[1].execution_date, 2, data)
@@ -1906,7 +1971,7 @@ def test_with_base_date_and_num_runs_and_execution_date_outside(self):
             ),
             data=dict(username='test', password='test'),
             follow_redirects=True,
-        )
+            )
         self.test.assertEqual(response.status_code, 200)
         data = response.data.decode('utf-8')
         self.assert_base_date_and_num_runs(self.runs[1].execution_date, 42, data)
@@ -1931,7 +1996,7 @@ def test_with_base_date_and_num_runs_and_execution_date_within(self):
             ),
             data=dict(username='test', password='test'),
             follow_redirects=True,
-        )
+            )
         self.test.assertEqual(response.status_code, 200)
         data = response.data.decode('utf-8')
         self.assert_base_date_and_num_runs(self.runs[2].execution_date, 5, data)
@@ -2922,7 +2987,6 @@ def test_start_date_filter(self):
 
 class TestRenderedView(TestBase):
     def setUp(self):
-
         self.default_date = datetime(2020, 3, 1)
         self.dag = DAG(
             "testdag",
@@ -3181,7 +3245,6 @@ def get_link(self, operator, dttm):  # pylint: disable=unused-argument
                 return 'https://airflow.apache.org'
 
         class DummyTestOperator(BaseOperator):
-
             operator_extra_links = (
                 RaiseErrorLink(),
                 NoResponseLink(),


Code review comments:
diff_hunk: @@ -270,6 +270,73 @@ def test_prefill_form_null_extra(self):
         cmv.prefill_form(form=mock_form, pk=1)
 
 
+class TestMultiDuplicateConnectionModelView(TestBase):
path: tests/www/test_views.py
text: @user1:
Please don't create a new test class -- instead add methods to the existing TestConnectionModelView

diff_hunk: @@ -3012,6 +3015,57 @@ def action_muldelete(self, items):
         self.update_redirect()
         return redirect(self.get_redirect())
 
+    @action(
+        'mulduplicate',
+        'Duplicate',
+        'Are you sure you want to duplicate the selected connections?',
+        single=False,
+    )
+    @provide_session
+    @auth.has_access(
+        [
+            (permissions.ACTION_CAN_CREATE, permissions.RESOURCE_CONNECTION),
+        ]
+    )
+    def action_mulduplicate(self, connections, session=None):
+        """Duplicate Multiple connections"""
+        for selected_conn in connections:
+            new_conn_id = selected_conn.conn_id
+            matches = re.findall(r"_Copy\((\d+)\)", selected_conn.conn_id)
path: airflow/www/views.py
text: @user1:
Is it intended to find `_CopyX` *anywhere*? Since the logic below only always adds a suffix, I would expect this to only check for a suffix as well.

@author:
This is a good point,
In case of something like **abc_copy1_xyz**,  we will currently end up doing **abc_copy2_xyz**, but **abc_copy1_xyz_copy1** should be the accepted behaviour.

diff_hunk: @@ -270,6 +270,73 @@ def test_prefill_form_null_extra(self):
         cmv.prefill_form(form=mock_form, pk=1)
 
 
+class TestMultiDuplicateConnectionModelView(TestBase):
+    def setUp(self):
+        super().setUp()
+        conn1 = Connection(
+            conn_id='google_cloud_default',
+            conn_type='Google Cloud',
+            description='Default Google Cloud Connection',
+        )
+
+        # Already copied and modified
+        conn2 = Connection(
+            conn_id='mongodb',
+            conn_type='FTP',
+            description='MongoDB2',
+            host='localhost',
+            schema='airflow',
+            port='5567',
+        )
+
+        conn3 = Connection(
+            conn_id='mysql_connection',
+            conn_type='FTP',
+            description='MongoDB2',
+            host='localhost',
+            schema='airflow',
+            port='3306',
+        )
+
+        conn4 = Connection(
+            conn_id='postgres_connection_Copy(1)',
+            conn_type='FTP',
+            description='Postgres',
+            host='localhost',
+            schema='airflow',
+            port='3306',
+        )
+
+        self.clear_table(Connection)
+        self.session.add_all([conn1, conn2, conn3, conn4])
+        self.session.commit()
+
+    def tearDown(self):
+        self.clear_table(Connection)
+        super().tearDown()
+
+    def test_duplicate_connection(self):
+        """ Test Duplicate multiple connection with suffix"""
+
+        mock_form = mock.Mock()
+        mock_form.data = {"action": "mulduplicate", "rowid": [1, 2, 4]}
+        resp = self.client.post('/connection/action_post', data=mock_form.data, follow_redirects=True)
+
+        expected_result = {
+            'google_cloud_default',
+            'google_cloud_default_Copy(1)',
+            'mongodb',
+            'mongodb_Copy(1)',
+            'mysql_connection',
+            'postgres_connection_Copy(1)',
+            'postgres_connection_Copy(2)',
+        }
+        response = {conn[0].strip() for conn in self.session.query(Connection.conn_id).all()}
path: tests/www/test_views.py
text: @user1:
```suggestion
        response = {conn[0] for conn in self.session.query(Connection.conn_id).all()}
```

strip shouldn't be needed.