Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

APP-4781 | Handled deleted AtlanTagName gracefully in Purpose asset deserialization #483

Merged
merged 1 commit into from
Jan 21, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 2 additions & 6 deletions pyatlan/model/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ def _convert_to_display_text(cls, data):
if display_text := AtlanTagCache.get_name_for_id(data):
return AtlanTagName(display_text)
else:
raise ValueError(f"{data} is not a valid AtlanTag")
return cls.get_deleted_sentinel()

@staticmethod
def json_encode_atlan_tag(atlan_tag_name: "AtlanTagName"):
Expand Down Expand Up @@ -237,11 +237,7 @@ def source_tag_attachements(self) -> List[SourceTagAttachment]:
def type_name_is_tag_name(cls, value):
if isinstance(value, AtlanTagName):
return value
try:
value = AtlanTagName._convert_to_display_text(value)
except ValueError:
value = AtlanTagName.get_deleted_sentinel()
return value
return AtlanTagName._convert_to_display_text(value)

def __init__(self, *args, **kwargs):
from pyatlan.cache.atlan_tag_cache import AtlanTagCache
Expand Down
76 changes: 73 additions & 3 deletions tests/unit/test_atlan_tag_name.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,11 @@
# SPDX-License-Identifier: Apache-2.0
# Copyright 2022 Atlan Pte. Ltd.
import pytest
from pydantic.v1 import parse_obj_as

import pyatlan.cache.atlan_tag_cache
from pyatlan.model.assets import Purpose
from pyatlan.model.constants import DELETED_
from pyatlan.model.core import AtlanTagName

ATLAN_TAG_ID = "yiB7RLvdC2yeryLPjaDeHM"
Expand Down Expand Up @@ -62,7 +65,7 @@ def test_convert_to_display_text_when_atlan_tag_passed_returns_same_atlan_tag(
assert good_atlan_tag is AtlanTagName._convert_to_display_text(good_atlan_tag)


def test_convert_to_display_text_when_bad_string_raises_value_error(monkeypatch):
def test_convert_to_display_text_when_bad_string(monkeypatch):
def get_name_for_id(_):
return None

Expand All @@ -72,8 +75,10 @@ def get_name_for_id(_):
get_name_for_id,
)

with pytest.raises(ValueError, match="bad is not a valid AtlanTag"):
AtlanTagName._convert_to_display_text("bad")
assert (
AtlanTagName._convert_to_display_text("bad").__repr__()
== f"AtlanTagName('{DELETED_}')"
)


def test_convert_to_display_text_when_id(monkeypatch):
Expand Down Expand Up @@ -102,3 +107,68 @@ def get_id_for_name(value):

def test_json_encode_atlan_tag(monkeypatch, good_atlan_tag):
assert AtlanTagName.json_encode_atlan_tag(good_atlan_tag) == ATLAN_TAG_ID


def test_asset_tag_name_field_deserialization(monkeypatch):
def get_name_for_id(_):
return None

def get_id_for_name(_):
return None

monkeypatch.setattr(
pyatlan.cache.atlan_tag_cache.AtlanTagCache,
"get_id_for_name",
get_id_for_name,
)

monkeypatch.setattr(
pyatlan.cache.atlan_tag_cache.AtlanTagCache,
"get_name_for_id",
get_name_for_id,
)
# Simulate a `Purpose` asset with `purpose_atlan_tags` of type `AtlanTagName`
purpose_asset = {
"typeName": "Purpose",
"attributes": {
# AtlanTagName
"purposeClassifications": [
"some-deleted-purpose-tag-1",
"some-deleted-purpose-tag-2",
],
},
"guid": "9f7a35f4-8d37-4273-81ec-c497a83a2472",
"status": "ACTIVE",
"classifications": [
# AtlanTag
{
"typeName": "some-deleted-purpose-tag-1",
"entityGuid": "82683fb9-1501-4627-a5d0-0da9be64c0d5",
"entityStatus": "DELETED",
"propagate": False,
"removePropagationsOnEntityDelete": True,
"restrictPropagationThroughLineage": True,
"restrictPropagationThroughHierarchy": False,
},
{
"typeName": "some-deleted-purpose-tag-2",
"entityGuid": "82683fb9-1501-4627-a5d0-0da9be64c0d5",
"entityStatus": "DELETED",
"propagate": False,
"removePropagationsOnEntityDelete": True,
"restrictPropagationThroughLineage": True,
"restrictPropagationThroughHierarchy": False,
},
],
}
purpose = parse_obj_as(Purpose, purpose_asset)
assert purpose and isinstance(purpose, Purpose)

# Verify that deleted tags are correctly set to `None`
# assert purpose.atlan_tags == [AtlanTagName('(DELETED)')]
assert purpose.atlan_tags and len(purpose.atlan_tags) == 2
assert purpose.atlan_tags[0].type_name.__repr__() == f"AtlanTagName('{DELETED_}')"
assert purpose.atlan_tags[1].type_name.__repr__() == f"AtlanTagName('{DELETED_}')"
assert purpose.purpose_atlan_tags and len(purpose.purpose_atlan_tags) == 2
assert purpose.purpose_atlan_tags[0].__repr__() == f"AtlanTagName('{DELETED_}')"
assert purpose.purpose_atlan_tags[1].__repr__() == f"AtlanTagName('{DELETED_}')"
Loading