Updates for SDK v0.6.2

Signed-off-by: Christopher Grote <cmgrote@users.noreply.github.com>

Author: cmgrote

custom_metadata/update_cm_on_assets.py

@@ -6,7 +6,7 @@
 from pyatlan.client.atlan import AtlanClient
 from pyatlan.model.assets import Asset
 from pyatlan.model.enums import AtlanConnectorType
-from pyatlan.model.search import DSL, IndexSearchRequest, TermAttributes, Terms
+from pyatlan.model.fluent_search import FluentSearch
 from pyatlan.utils import get_logger
 
 CUSTOM_METADATA_NAME = "Quality Data"
@@ -19,7 +19,7 @@ def find_asset(
     connection_name: str,
     asset_name: str,
     attributes: Optional[list[str]] = None,
-) -> Asset:
+) -> Optional[Asset]:
     """
     Given a connector type and otherwise-qualified name (not including the
     connection portion of the qualified_name), finds and returns the asset in
@@ -39,14 +39,14 @@ def find_asset(
     qualified_names = []
     for connection in connections:
         qualified_names.append(f"{connection.qualified_name}/{asset_name}")
-    by_name = Terms(field=TermAttributes.QUALIFIED_NAME.value, values=qualified_names)
-    dsl = DSL(query=by_name)
-    search_request = IndexSearchRequest(
-        dsl=dsl,
-        attributes=attributes,
-    )
-    results = client.search(search_request)
-    return results.current_page()[0] if results else None
+    search_request = (
+        FluentSearch(_includes_on_results=attributes).where(
+            Asset.QUALIFIED_NAME.within(qualified_names)
+        )
+    ).to_request()
+    if results := client.search(search_request):
+        return results.current_page()[0]
+    return None
 
 
 def update_custom_metadata(
@@ -80,29 +80,33 @@ def update_custom_metadata(
 
 
 def main():
-    asset = find_asset(
+    if asset := find_asset(
         connector_type=AtlanConnectorType.SNOWFLAKE,
         connection_name="development",
         asset_name="RAW/WIDEWORLDIMPORTERS_PURCHASING/SUPPLIERS",
         attributes=CustomMetadataCache.get_attributes_for_search_results(
             CUSTOM_METADATA_NAME
         ),
-    )
-    logger.info(f"Found asset: {asset}")
-    updated = update_custom_metadata(
-        asset=asset,
-        rating="OK",
-        passed=10,
-        failed=5,
-        reports=["https://www.example.com", "https://www.atlan.com"],
-    )
-    # Note that the updated asset will NOT show the custom metadata, if you want
-    # to see the custom metadata you need to re-retrieve the asset itself
-    assert updated  # noqa: S101
-    result = client.get_asset_by_guid(
-        guid=updated.guid, asset_type=type(updated), ignore_relationships=True
-    )
-    logger.info(f"Asset's custom metadata was updated: {result}")
+    ):
+        logger.info(f"Found asset: {asset}")
+        updated = update_custom_metadata(
+            asset=asset,
+            rating="OK",
+            passed=10,
+            failed=5,
+            reports=["https://www.example.com", "https://www.atlan.com"],
+        )
+        # Note that the updated asset will NOT show the custom metadata, if you want
+        # to see the custom metadata you need to re-retrieve the asset itself
+        assert updated  # noqa: S101
+        result = client.get_asset_by_guid(
+            guid=updated.guid, asset_type=type(updated), ignore_relationships=True
+        )
+        logger.info(f"Asset's custom metadata was updated: {result}")
+    else:
+        logger.warn(
+            "Unable to find asset: (development)/RAW/WIDEWORLDIMPORTERS_PURCHASING/SUPPLIERS"
+        )
 
 
 if __name__ == "__main__":

requirements.txt

@@ -1 +1 @@
-pyatlan>=0.4.0
+pyatlan>=0.6.2

search/and_star_assets.py

@@ -2,9 +2,9 @@
 # Copyright 2023 Atlan Pte. Ltd.
 import time
 
-from pyatlan.client.atlan import AtlanClient, IndexSearchRequest
+from pyatlan.client.atlan import AtlanClient
 from pyatlan.model.assets import Asset, AtlasGlossaryTerm
-from pyatlan.model.search import DSL, Term
+from pyatlan.model.fluent_search import FluentSearch
 from pyatlan.model.structs import StarredDetails
 from pyatlan.utils import get_logger
 
@@ -20,14 +20,17 @@ def find_assets() -> AtlanClient.SearchResults:
 
     :returns: results of the search
     """
-    are_active = Term.with_state("ACTIVE")
-    are_term = Term.with_type_name("AtlasGlossaryTerm")
     glossary = client.find_glossary_by_name("Metrics")
-    in_glossary = Term.with_glossary(glossary.qualified_name)
-    dsl = DSL(query=are_active + are_term + in_glossary, from_=0, size=100)
-    search_request = IndexSearchRequest(
-        dsl=dsl, attributes=["starredDetailsList", "starredBy", "anchor"]
-    )
+    search_request = (
+        FluentSearch()
+        .where(FluentSearch.active_assets())
+        .where(FluentSearch.asset_type(AtlasGlossaryTerm))
+        .where(AtlasGlossaryTerm.ANCHOR.eq(glossary.qualified_name))
+        .page_size(100)
+        .include_on_results(Asset.STARRED_DETAILS_LIST)
+        .include_on_results(Asset.STARRED_BY)
+        .include_on_results(AtlasGlossaryTerm.ANCHOR)
+    ).to_request()
     return client.search(search_request)
 
 

search/and_traverse_lineage.py

@@ -1,28 +1,34 @@
 # SPDX-License-Identifier: Apache-2.0
 # Copyright 2023 Atlan Pte. Ltd.
-from pyatlan.client.atlan import AtlanClient, IndexSearchRequest
+from pyatlan.client.atlan import AtlanClient
 from pyatlan.model.assets import Asset, SigmaWorkbook
-from pyatlan.model.enums import AtlanComparisonOperator, LineageDirection
+from pyatlan.model.enums import (
+    AtlanComparisonOperator,
+    CertificateStatus,
+    LineageDirection,
+)
+from pyatlan.model.fluent_search import FluentSearch
 from pyatlan.model.lineage import EntityFilter, FilterList, LineageListRequest
-from pyatlan.model.search import DSL, Term
 from pyatlan.utils import get_logger
 
 client = AtlanClient()
 logger = get_logger(level="INFO")
 
 
-def find_all(type_name: str) -> AtlanClient.SearchResults:
+def find_all(asset_type: type) -> AtlanClient.SearchResults:
     """
     This query will find all assets of the specified type
     that are active (not archived or soft-deleted).
 
-    :param type_name: type of assets to find
+    :param asset_type: type of assets to find
     :returns: results of the search
     """
-    are_active = Term.with_state("ACTIVE")
-    are_term = Term.with_type_name(type_name)
-    dsl = DSL(query=are_active + are_term, from_=0, size=100)
-    search_request = IndexSearchRequest(dsl=dsl)
+    search_request = (
+        FluentSearch()
+        .where(FluentSearch.asset_type(asset_type))
+        .where(FluentSearch.active_assets())
+        .page_size(100)
+    ).to_request()
     return client.search(search_request)
 
 
@@ -41,14 +47,17 @@ def upstream_certified_sources(guid: str) -> list[Asset]:
     request.direction = LineageDirection.UPSTREAM
     request.offset = 0
     request.size = 100
-    request.attributes = ["name", "certificateStatus"]
+    request.attributes = [
+        Asset.NAME.atlan_field_name,
+        Asset.CERTIFICATE_STATUS.atlan_field_name,
+    ]
     request.entity_filters = FilterList(
         condition="AND",
         criteria=[
             EntityFilter(
-                attribute_name="certificateStatus",
+                attribute_name=Asset.CERTIFICATE_STATUS.atlan_field_name,
                 operator=AtlanComparisonOperator.CONTAINS,
-                attribute_value="VERIFIED",
+                attribute_value=CertificateStatus.VERIFIED.value,
             )
         ],
     )
@@ -61,7 +70,7 @@ def upstream_certified_sources(guid: str) -> list[Asset]:
 
 
 def main():
-    results = find_all(type_name="SigmaWorkbook")
+    results = find_all(SigmaWorkbook)
     for workbook in results:
         if isinstance(workbook, SigmaWorkbook):
             verified_sources = upstream_certified_sources(workbook.guid)