Duplicate Connection: Added logic to query if a connection id exists before creating one

[subkanthi]

Added logic to query if a connection exists before trying to create one in the duplicate connection flow.
closes: #18050

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.

[potiuk]

LGTM - but how about adding a test for that one ?

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest main or amend the last commit of the PR, and push it with --force-with-lease.

[subkanthi]

LGTM - but how about adding a test for that one ?

Thanks , sure will do.

[potiuk]

OK. Ping us when there are tests :)

[subkanthi]

OK. Ping us when there are tests :)

There is a test in test_views_connection.py that tests the duplication logic, I can add one more to test the logic of erroring out when 10 connections already exists.

`def test_duplicate_connection(admin_client):
"""Test Duplicate multiple connection with suffix"""
conn1 = Connection(
conn_id='test_duplicate_gcp_connection',
conn_type='Google Cloud',
description='Google Cloud Connection',
)
conn2 = Connection(
conn_id='test_duplicate_mysql_connection',
conn_type='FTP',
description='MongoDB2',
host='localhost',
schema='airflow',
port=3306,
)
conn3 = Connection(
conn_id='test_duplicate_postgres_connection_copy1',
conn_type='FTP',
description='Postgres',
host='localhost',
schema='airflow',
port=3306,
)
with create_session() as session:
session.query(Connection).delete()
session.add_all([conn1, conn2, conn3])
session.commit()

data = {"action": "mulduplicate", "rowid": [conn1.id, conn3.id]}
resp = admin_client.post('/connection/action_post', data=data, follow_redirects=True)
expected_result = {
    'test_duplicate_gcp_connection',
    'test_duplicate_gcp_connection_copy1',
    'test_duplicate_mysql_connection',
    'test_duplicate_postgres_connection_copy1',
    'test_duplicate_postgres_connection_copy2',
}
response = {conn[0] for conn in session.query(Connection.conn_id).all()}
assert resp.status_code == 200
assert expected_result == response`

[potiuk]

There is a test in test_views_connection.py that tests the duplication logic, I can add one more to test the logic of erroring out when 10 connections already exists.

sounds good :)

[potiuk]

Wooho!