Merge pull request #45 from atlanhq/find_connections

Add find_connections_by_name method to AtlanClient

Author: ErnestoLoma

HISTORY.md

@@ -1,3 +1,7 @@
+## 0.0.28 (May 9, 2023)
+
+* Add find_connections_by_name to AtlanClient
+
 ## 0.0.27 (May 8, 2023)
 
 * Add a create method to Process

pyatlan/client/atlan.py

@@ -61,10 +61,15 @@
     CustomMetadata,
     CustomMetadataReqest,
 )
-from pyatlan.model.enums import AtlanDeleteType, AtlanTypeCategory, CertificateStatus
+from pyatlan.model.enums import (
+    AtlanConnectorType,
+    AtlanDeleteType,
+    AtlanTypeCategory,
+    CertificateStatus,
+)
 from pyatlan.model.response import AssetMutationResponse
 from pyatlan.model.role import RoleResponse
-from pyatlan.model.search import IndexSearchRequest
+from pyatlan.model.search import DSL, IndexSearchRequest, Term
 from pyatlan.model.typedef import (
     ClassificationDef,
     CustomMetadataDef,
@@ -702,3 +707,24 @@ def remove_terms(
         if assets := response.assets_updated(asset_type=asset_type):
             return assets[0]
         return asset
+
+    @validate_arguments()
+    def find_connections_by_name(
+        self,
+        name: str,
+        connector_type: AtlanConnectorType,
+        attributes: list[str] = None,
+    ) -> list[Connection]:
+        query = (
+            Term.with_state("ACTIVE")
+            + Term.with_type_name("CONNECTION")
+            + Term.with_name(name)
+            + Term(field="connectorName", value=connector_type.value)
+        )
+        dsl = DSL(query=query)
+        search_request = IndexSearchRequest(
+            dsl=dsl,
+            attributes=attributes,
+        )
+        results = self.search(search_request)
+        return [asset for asset in results if isinstance(asset, Connection)]

pyatlan/version.txt

@@ -1 +1 @@
-0.0.27
+0.0.28

tests/integration/test_client.py

@@ -5,6 +5,7 @@
 
 from pyatlan.client.atlan import AtlanClient
 from pyatlan.model.assets import AtlasGlossary, AtlasGlossaryTerm, Connection, Database
+from pyatlan.model.enums import AtlanConnectorType
 
 iter_count = count(1)
 
@@ -219,3 +220,13 @@ def test_remove_term(
     assert deleted_terms[0].guid == original_term.guid
     active_terms = [t for t in database.terms if t.relationship_status != "DELETED"]
     assert active_terms[0].guid == another_term.guid
+
+
+def test_find_connections_by_name(client: AtlanClient):
+    connections = client.find_connections_by_name(
+        name="Test Connection",
+        connector_type=AtlanConnectorType.SNOWFLAKE,
+        attributes=["connectorName"],
+    )
+    assert len(connections) == 1
+    assert connections[0].connector_name == AtlanConnectorType.SNOWFLAKE.value