Skip to content

Commit ee73c28

Browse files
committed
Add create method to Badge
1 parent f56ec3c commit ee73c28

6 files changed

Lines changed: 3032 additions & 622 deletions

File tree

pyatlan/cache/custom_metadata_cache.py

Lines changed: 25 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
from typing import Any, Optional
55

66
from pyatlan.client.atlan import AtlanClient
7-
from pyatlan.error import LogicError, NotFoundError
7+
from pyatlan.error import InvalidRequestError, LogicError, NotFoundError
88
from pyatlan.model.core import CustomMetadata
99
from pyatlan.model.enums import AtlanTypeCategory
1010
from pyatlan.model.typedef import AttributeDef, CustomMetadataDef
@@ -95,22 +95,44 @@ def get_id_for_name(cls, name: str) -> Optional[str]:
9595
"""
9696
Translate the provided human-readable custom metadata set name to its Atlan-internal ID string.
9797
"""
98+
if name is None or not name.strip():
99+
raise InvalidRequestError(
100+
message="No name was provided when attempting to retrieve custom metadata.",
101+
code="ATLAN-PYTHON-404-008",
102+
param="",
103+
)
98104
if cm_id := cls.map_name_to_id.get(name):
99105
return cm_id
100106
# If not found, refresh the cache and look again (could be stale)
101107
cls.refresh_cache()
102-
return cls.map_name_to_id.get(name)
108+
if cm_id := cls.map_name_to_id.get(name):
109+
return cm_id
110+
raise NotFoundError(
111+
message=f"Custom metadata with name {name} does not exist.",
112+
code="ATLAN-PYTHON-404-009",
113+
)
103114

104115
@classmethod
105116
def get_name_for_id(cls, idstr: str) -> Optional[str]:
106117
"""
107118
Translate the provided Atlan-internal custom metadata ID string to the human-readable custom metadata set name.
108119
"""
120+
if idstr is None or not idstr.strip():
121+
raise InvalidRequestError(
122+
message="No ID was provided when attempting to retrieve custom metadata.",
123+
code="ATLAN-PYTHON-404-008",
124+
param="",
125+
)
109126
if cm_name := cls.map_id_to_name.get(idstr):
110127
return cm_name
111128
# If not found, refresh the cache and look again (could be stale)
112129
cls.refresh_cache()
113-
return cls.map_id_to_name.get(idstr)
130+
if cm_name := cls.map_id_to_name.get(idstr):
131+
return cm_name
132+
raise NotFoundError(
133+
message=f"Custom metadata with ID {idstr} does not exist.",
134+
code="ATLAN-PYTHON-404-009",
135+
)
114136

115137
@classmethod
116138
def get_type_for_id(cls, idstr: str) -> Optional[type]:

pyatlan/exceptions.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ def __init__(
4848
code: Optional[str],
4949
cause: Optional[Exception] = None,
5050
):
51+
super().__init__(message)
5152
self.message = message
5253
self.code = code
5354
self.status_code = status_code

pyatlan/generator/templates/entity.jinja2

Lines changed: 34 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,10 @@ from pyatlan.model.structs import (
5050
GoogleTag,
5151
Histogram,
5252
KafkaTopicConsumption,
53+
MCRuleComparison,
54+
MCRuleSchedule,
5355
PopularityInsights,
56+
SourceTagAttribute,
5457
)
5558
from pyatlan.utils import next_id
5659
from urllib.parse import quote, unquote
@@ -361,7 +364,17 @@ class {{ entity_def.name }}({{super_classes[0]}} {%- if "Asset" in super_classes
361364
raise ValueError('must be {{ entity_def.name }}')
362365
return v
363366

364-
{%- if entity_def.name == "Readme" %}
367+
{%- if entity_def.name == "Badge" %}
368+
@classmethod
369+
# @validate_arguments()
370+
def create(cls, *, name: StrictStr, cm_name: str, cm_attribute: str) -> Badge:
371+
validate_required_fields(["name", "cm_name", "cm_attribute"], [name])
372+
return cls(
373+
attributes=Badge.Attributes.create(
374+
name=name, cm_name=cm_name, cm_attribute=cm_attribute
375+
)
376+
)
377+
{%- elif entity_def.name == "Readme" %}
365378
@classmethod
366379
# @validate_arguments()
367380
def create(
@@ -523,6 +536,26 @@ class {{ entity_def.name }}({{super_classes[0]}} {%- if "Asset" in super_classes
523536
inputs=inputs,
524537
outputs=outputs,
525538
)
539+
{%- elif entity_def.name == "Badge" %}
540+
@classmethod
541+
# @validate_arguments()
542+
def create(
543+
cls, *, name: StrictStr, cm_name: str, cm_attribute: str
544+
) -> Badge.Attributes:
545+
validate_required_fields(
546+
["name", "cm_name", "cm_attribute"], [name, cm_name, cm_attribute]
547+
)
548+
from pyatlan.cache.custom_metadata_cache import CustomMetadataCache
549+
550+
cm_id = CustomMetadataCache.get_id_for_name(cm_name)
551+
cm_attr_id = CustomMetadataCache.get_attr_id_for_name(
552+
set_name=cm_name, attr_name=cm_attribute
553+
)
554+
return Badge.Attributes(
555+
name=name,
556+
qualified_name=f"badges/global/{cm_id}.{cm_attr_id}",
557+
badge_metadata_attribute=f"{cm_id}.{cm_attr_id}",
558+
)
526559
{%- elif entity_def.name == "Readme" %}
527560
@classmethod
528561
# @validate_arguments()

0 commit comments

Comments
 (0)