From 7a32b1c35fb3ad5556623329be0fb161d8a46f98 Mon Sep 17 00:00:00 2001 From: Angad Sethi Date: Thu, 3 Sep 2026 12:56:43 +0530 Subject: [PATCH] feat: add Notification asset model and related enums --- pyatlan/generator/templates/imports.jinja2 | 1 + pyatlan/generator/templates/structs.jinja2 | 1 + pyatlan/model/assets/__init__.py | 4 +- pyatlan/model/assets/__init__.pyi | 7 +- pyatlan/model/assets/core/__init__.py | 6 +- pyatlan/model/assets/core/asset.py | 8 +- .../model/assets/core/atlan_app_workflow.py | 31 + pyatlan/model/assets/core/notification.py | 251 ++++++ .../assets/core/workflow_notification.py | 227 ++++++ pyatlan/model/enums.py | 27 + pyatlan/model/structs.py | 20 + pyatlan_v9/model/assets/__init__.py | 5 + pyatlan_v9/model/assets/__init__.pyi | 10 + pyatlan_v9/model/assets/_init_notification.py | 17 + .../assets/_init_workflow_notification.py | 17 + pyatlan_v9/model/assets/atlan_app_workflow.py | 16 + pyatlan_v9/model/assets/notification.py | 646 +++++++++++++++ .../model/assets/notification_related.py | 64 ++ .../model/assets/workflow_notification.py | 767 ++++++++++++++++++ .../assets/workflow_notification_related.py | 53 ++ 20 files changed, 2170 insertions(+), 8 deletions(-) create mode 100644 pyatlan/model/assets/core/notification.py create mode 100644 pyatlan/model/assets/core/workflow_notification.py create mode 100644 pyatlan_v9/model/assets/_init_notification.py create mode 100644 pyatlan_v9/model/assets/_init_workflow_notification.py create mode 100644 pyatlan_v9/model/assets/notification.py create mode 100644 pyatlan_v9/model/assets/notification_related.py create mode 100644 pyatlan_v9/model/assets/workflow_notification.py create mode 100644 pyatlan_v9/model/assets/workflow_notification_related.py diff --git a/pyatlan/generator/templates/imports.jinja2 b/pyatlan/generator/templates/imports.jinja2 index 5491ee2ca..823bce1b9 100644 --- a/pyatlan/generator/templates/imports.jinja2 +++ b/pyatlan/generator/templates/imports.jinja2 @@ -140,6 +140,7 @@ from pyatlan.model.structs import ( AssetExternalDQTestsByStatus, AwsCloudWatchMetric, DataQualityRuleThresholdObject, + NotificationExternalReference, SQLProcedureAdditionalProperties, SQLProcedureReturnType, SageMakerUnifiedStudioAssetFilter, diff --git a/pyatlan/generator/templates/structs.jinja2 b/pyatlan/generator/templates/structs.jinja2 index 1bbdd9394..dd14e1a8b 100644 --- a/pyatlan/generator/templates/structs.jinja2 +++ b/pyatlan/generator/templates/structs.jinja2 @@ -17,6 +17,7 @@ from pyatlan.model.enums import ( DataQualityRuleThresholdUnit, AppWorkflowRunStatus, AssetSmusMetadataFormStatus, + NotificationSurface, ) from pyatlan.model.utils import to_camel_case from pyatlan.utils import select_optional_set_fields, validate_required_fields diff --git a/pyatlan/model/assets/__init__.py b/pyatlan/model/assets/__init__.py index 389316906..ebe0a6905 100644 --- a/pyatlan/model/assets/__init__.py +++ b/pyatlan/model/assets/__init__.py @@ -9,6 +9,7 @@ "Process", "AtlasGlossaryCategory", "AccessControl", + "Notification", "AuthPolicy", "StakeholderTitle", "Catalog", @@ -23,6 +24,7 @@ "V1CoalesceProcess", "ColumnProcess", "Persona", + "WorkflowNotification", "App", "Airflow", "ADF", @@ -117,8 +119,8 @@ "Snowflake", "DatabricksUnityCatalogTag", "SnowflakeStream", - "Database", "CalculationView", + "Database", "Procedure", "Table", "SnowflakePipe", diff --git a/pyatlan/model/assets/__init__.pyi b/pyatlan/model/assets/__init__.pyi index 1ea35142e..f81a01fbb 100644 --- a/pyatlan/model/assets/__init__.pyi +++ b/pyatlan/model/assets/__init__.pyi @@ -6,6 +6,7 @@ __all__ = [ "Process", "AtlasGlossaryCategory", "AccessControl", + "Notification", "AuthPolicy", "StakeholderTitle", "Catalog", @@ -20,6 +21,7 @@ __all__ = [ "V1CoalesceProcess", "ColumnProcess", "Persona", + "WorkflowNotification", "App", "Airflow", "ADF", @@ -114,8 +116,8 @@ __all__ = [ "Snowflake", "DatabricksUnityCatalogTag", "SnowflakeStream", - "Database", "CalculationView", + "Database", "Procedure", "Table", "SnowflakePipe", @@ -234,7 +236,6 @@ __all__ = [ "Unstructured", "V1Coalesce", "EventStore", - "NoSQL", "Insight", "AssetGrouping", "ObjectStore", @@ -724,6 +725,7 @@ from .core.mongo_d_b_database import MongoDBDatabase from .core.monte_carlo import MonteCarlo from .core.namespace import Namespace from .core.no_s_q_l import NoSQL +from .core.notification import Notification from .core.partial import Partial from .core.partial_field import PartialField from .core.partial_object import PartialObject @@ -796,6 +798,7 @@ from .core.table_partition import TablePartition from .core.tag import Tag from .core.v1_coalesce_process import V1CoalesceProcess from .core.view import View +from .core.workflow_notification import WorkflowNotification from .cube import Cube from .cube_dimension import CubeDimension from .cube_field import CubeField diff --git a/pyatlan/model/assets/core/__init__.py b/pyatlan/model/assets/core/__init__.py index b6c332ea7..23ab073e4 100644 --- a/pyatlan/model/assets/core/__init__.py +++ b/pyatlan/model/assets/core/__init__.py @@ -140,6 +140,7 @@ from .monte_carlo import MonteCarlo from .namespace import Namespace from .no_s_q_l import NoSQL +from .notification import Notification from .partial import Partial from .partial_field import PartialField from .partial_object import PartialObject @@ -211,6 +212,7 @@ from .tag import Tag from .v1_coalesce_process import V1CoalesceProcess from .view import View +from .workflow_notification import WorkflowNotification # Update asset forward references: localns = locals() @@ -222,6 +224,7 @@ Process.Attributes.update_forward_refs(**localns) AtlasGlossaryCategory.Attributes.update_forward_refs(**localns) AccessControl.Attributes.update_forward_refs(**localns) +Notification.Attributes.update_forward_refs(**localns) AuthPolicy.Attributes.update_forward_refs(**localns) StakeholderTitle.Attributes.update_forward_refs(**localns) Catalog.Attributes.update_forward_refs(**localns) @@ -236,6 +239,7 @@ V1CoalesceProcess.Attributes.update_forward_refs(**localns) ColumnProcess.Attributes.update_forward_refs(**localns) Persona.Attributes.update_forward_refs(**localns) +WorkflowNotification.Attributes.update_forward_refs(**localns) App.Attributes.update_forward_refs(**localns) Airflow.Attributes.update_forward_refs(**localns) ADF.Attributes.update_forward_refs(**localns) @@ -329,8 +333,8 @@ Snowflake.Attributes.update_forward_refs(**localns) DatabricksUnityCatalogTag.Attributes.update_forward_refs(**localns) SnowflakeStream.Attributes.update_forward_refs(**localns) -Database.Attributes.update_forward_refs(**localns) CalculationView.Attributes.update_forward_refs(**localns) +Database.Attributes.update_forward_refs(**localns) Procedure.Attributes.update_forward_refs(**localns) Table.Attributes.update_forward_refs(**localns) SnowflakePipe.Attributes.update_forward_refs(**localns) diff --git a/pyatlan/model/assets/core/asset.py b/pyatlan/model/assets/core/asset.py index df4e3fb1b..1e5bf9363 100644 --- a/pyatlan/model/assets/core/asset.py +++ b/pyatlan/model/assets/core/asset.py @@ -1063,13 +1063,13 @@ def __setattr__(self, name, value): "assetSodaLastSyncRunAt", "assetSodaLastSyncRunAt" ) """ - + """ ASSET_SODA_LAST_SCAN_AT: ClassVar[NumericField] = NumericField( "assetSodaLastScanAt", "assetSodaLastScanAt" ) """ - + """ ASSET_SODA_CHECK_STATUSES: ClassVar[TextField] = TextField( "assetSodaCheckStatuses", "assetSodaCheckStatuses" @@ -1081,7 +1081,7 @@ def __setattr__(self, name, value): "assetSodaSourceURL", "assetSodaSourceURL" ) """ - + """ ASSET_ICON: ClassVar[TextField] = TextField("assetIcon", "assetIcon") """ @@ -1147,7 +1147,7 @@ def __setattr__(self, name, value): "isAIGenerated", "isAIGenerated" ) """ - + """ ASSET_COVER_IMAGE: ClassVar[TextField] = TextField( "assetCoverImage", "assetCoverImage" diff --git a/pyatlan/model/assets/core/atlan_app_workflow.py b/pyatlan/model/assets/core/atlan_app_workflow.py index 4ed0d5e28..c11999b92 100644 --- a/pyatlan/model/assets/core/atlan_app_workflow.py +++ b/pyatlan/model/assets/core/atlan_app_workflow.py @@ -107,6 +107,12 @@ def __setattr__(self, name, value): """ TBC """ + WORKFLOW_NOTIFICATION_NOTIFICATIONS: ClassVar[RelationField] = RelationField( + "workflowNotificationNotifications" + ) + """ + TBC + """ ATLAN_APP_WORKFLOW_RUNS: ClassVar[RelationField] = RelationField( "atlanAppWorkflowRuns" ) @@ -127,6 +133,7 @@ def __setattr__(self, name, value): "atlan_app_workflow_deployment_name", "atlan_app_workflow_runtime_mode", "atlan_app", + "workflow_notification_notifications", "atlan_app_workflow_runs", ] @@ -310,6 +317,26 @@ def atlan_app(self, atlan_app: Optional[AtlanApp]): self.attributes = self.Attributes() self.attributes.atlan_app = atlan_app + @property + def workflow_notification_notifications( + self, + ) -> Optional[List[WorkflowNotification]]: + return ( + None + if self.attributes is None + else self.attributes.workflow_notification_notifications + ) + + @workflow_notification_notifications.setter + def workflow_notification_notifications( + self, workflow_notification_notifications: Optional[List[WorkflowNotification]] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.workflow_notification_notifications = ( + workflow_notification_notifications + ) + @property def atlan_app_workflow_runs(self) -> Optional[List[AppWorkflowRun]]: return ( @@ -353,6 +380,9 @@ class Attributes(AtlanApp.Attributes): atlan_app: Optional[AtlanApp] = Field( default=None, description="" ) # relationship + workflow_notification_notifications: Optional[List[WorkflowNotification]] = ( + Field(default=None, description="") + ) # relationship atlan_app_workflow_runs: Optional[List[AppWorkflowRun]] = Field( default=None, description="" ) # relationship @@ -369,3 +399,4 @@ class Attributes(AtlanApp.Attributes): from .app_workflow_run import AppWorkflowRun # noqa: E402, F401 from .atlan_app import AtlanApp # noqa: E402, F401 +from .workflow_notification import WorkflowNotification # noqa: E402, F401 diff --git a/pyatlan/model/assets/core/notification.py b/pyatlan/model/assets/core/notification.py new file mode 100644 index 000000000..ec044fc04 --- /dev/null +++ b/pyatlan/model/assets/core/notification.py @@ -0,0 +1,251 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2025 Atlan Pte. Ltd. + + +from __future__ import annotations + +from datetime import datetime +from typing import ClassVar, List, Optional, Set + +from pydantic.v1 import Field, validator + +from pyatlan.model.enums import NotificationState +from pyatlan.model.fields.atlan_fields import KeywordField, NumericField, TextField +from pyatlan.model.structs import NotificationExternalReference + +from .asset import Asset + + +class Notification(Asset, type_name="Notification"): + """Description""" + + type_name: str = Field(default="Notification", allow_mutation=False) + + @validator("type_name") + def validate_type_name(cls, v): + if v != "Notification": + raise ValueError("must be Notification") + return v + + def __setattr__(self, name, value): + if name in Notification._convenience_properties: + return object.__setattr__(self, name, value) + super().__setattr__(name, value) + + NOTIFICATION_STATE: ClassVar[KeywordField] = KeywordField( + "notificationState", "notificationState" + ) + """ + Where the notification is in its lifecycle. Also how a check decides whether a live notification already exists before raising another. + """ # noqa: E501 + NOTIFICATION_RECIPIENT_USERS: ClassVar[KeywordField] = KeywordField( + "notificationRecipientUsers", "notificationRecipientUsers" + ) + """ + Usernames the notification was addressed to. Drives the assigned-to-me view together with the group and role recipients. + """ # noqa: E501 + NOTIFICATION_RECIPIENT_GROUPS: ClassVar[KeywordField] = KeywordField( + "notificationRecipientGroups", "notificationRecipientGroups" + ) + """ + Identifiers of the Atlan groups the notification is meant for. A group outlives its members, so a rule addressing a group stays generic as people join and leave. + """ # noqa: E501 + NOTIFICATION_RECIPIENT_ROLES: ClassVar[KeywordField] = KeywordField( + "notificationRecipientRoles", "notificationRecipientRoles" + ) + """ + Identifiers of the Atlan roles that need to see the notification. Resolved to individuals by the interface at read time rather than stored. + """ # noqa: E501 + NOTIFICATION_LAST_ACTED_BY: ClassVar[KeywordField] = KeywordField( + "notificationLastActedBy", "notificationLastActedBy" + ) + """ + Username of the person who last moved the state. Declared separately because the write is service-mediated, so the entity's own modifier records the service and not the person. + """ # noqa: E501 + NOTIFICATION_LAST_ACTED_AT: ClassVar[NumericField] = NumericField( + "notificationLastActedAt", "notificationLastActedAt" + ) + """ + When a person last moved the state. Read against the creation timestamp this gives time to resolution. + """ + NOTIFICATION_LAST_ACTED_REASON: ClassVar[TextField] = TextField( + "notificationLastActedReason", "notificationLastActedReason" + ) + """ + Reason or note the person gave for the last action. Mandatory for some actions and optional for others, so it is free text rather than a code. + """ # noqa: E501 + NOTIFICATION_EXTERNAL_REFERENCES: ClassVar[KeywordField] = KeywordField( + "notificationExternalReferences", "notificationExternalReferences" + ) + """ + Where this notification also lives outside Atlan, one entry per surface. This is what lets an action taken in the inbox go back and update the message it came from. + """ # noqa: E501 + + _convenience_properties: ClassVar[List[str]] = [ + "notification_state", + "notification_recipient_users", + "notification_recipient_groups", + "notification_recipient_roles", + "notification_last_acted_by", + "notification_last_acted_at", + "notification_last_acted_reason", + "notification_external_references", + ] + + @property + def notification_state(self) -> Optional[NotificationState]: + return None if self.attributes is None else self.attributes.notification_state + + @notification_state.setter + def notification_state(self, notification_state: Optional[NotificationState]): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.notification_state = notification_state + + @property + def notification_recipient_users(self) -> Optional[Set[str]]: + return ( + None + if self.attributes is None + else self.attributes.notification_recipient_users + ) + + @notification_recipient_users.setter + def notification_recipient_users( + self, notification_recipient_users: Optional[Set[str]] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.notification_recipient_users = notification_recipient_users + + @property + def notification_recipient_groups(self) -> Optional[Set[str]]: + return ( + None + if self.attributes is None + else self.attributes.notification_recipient_groups + ) + + @notification_recipient_groups.setter + def notification_recipient_groups( + self, notification_recipient_groups: Optional[Set[str]] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.notification_recipient_groups = notification_recipient_groups + + @property + def notification_recipient_roles(self) -> Optional[Set[str]]: + return ( + None + if self.attributes is None + else self.attributes.notification_recipient_roles + ) + + @notification_recipient_roles.setter + def notification_recipient_roles( + self, notification_recipient_roles: Optional[Set[str]] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.notification_recipient_roles = notification_recipient_roles + + @property + def notification_last_acted_by(self) -> Optional[str]: + return ( + None + if self.attributes is None + else self.attributes.notification_last_acted_by + ) + + @notification_last_acted_by.setter + def notification_last_acted_by(self, notification_last_acted_by: Optional[str]): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.notification_last_acted_by = notification_last_acted_by + + @property + def notification_last_acted_at(self) -> Optional[datetime]: + return ( + None + if self.attributes is None + else self.attributes.notification_last_acted_at + ) + + @notification_last_acted_at.setter + def notification_last_acted_at( + self, notification_last_acted_at: Optional[datetime] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.notification_last_acted_at = notification_last_acted_at + + @property + def notification_last_acted_reason(self) -> Optional[str]: + return ( + None + if self.attributes is None + else self.attributes.notification_last_acted_reason + ) + + @notification_last_acted_reason.setter + def notification_last_acted_reason( + self, notification_last_acted_reason: Optional[str] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.notification_last_acted_reason = notification_last_acted_reason + + @property + def notification_external_references( + self, + ) -> Optional[List[NotificationExternalReference]]: + return ( + None + if self.attributes is None + else self.attributes.notification_external_references + ) + + @notification_external_references.setter + def notification_external_references( + self, + notification_external_references: Optional[List[NotificationExternalReference]], + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.notification_external_references = ( + notification_external_references + ) + + class Attributes(Asset.Attributes): + notification_state: Optional[NotificationState] = Field( + default=None, description="" + ) + notification_recipient_users: Optional[Set[str]] = Field( + default=None, description="" + ) + notification_recipient_groups: Optional[Set[str]] = Field( + default=None, description="" + ) + notification_recipient_roles: Optional[Set[str]] = Field( + default=None, description="" + ) + notification_last_acted_by: Optional[str] = Field(default=None, description="") + notification_last_acted_at: Optional[datetime] = Field( + default=None, description="" + ) + notification_last_acted_reason: Optional[str] = Field( + default=None, description="" + ) + notification_external_references: Optional[ + List[NotificationExternalReference] + ] = Field(default=None, description="") + + attributes: Notification.Attributes = Field( + default_factory=lambda: Notification.Attributes(), + description=( + "Map of attributes in the instance and their values. " + "The specific keys of this map will vary by type, " + "so are described in the sub-types of this schema." + ), + ) diff --git a/pyatlan/model/assets/core/workflow_notification.py b/pyatlan/model/assets/core/workflow_notification.py new file mode 100644 index 000000000..5fab73626 --- /dev/null +++ b/pyatlan/model/assets/core/workflow_notification.py @@ -0,0 +1,227 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2025 Atlan Pte. Ltd. + + +from __future__ import annotations + +from datetime import datetime +from typing import ClassVar, List, Optional + +from pydantic.v1 import Field, validator + +from pyatlan.model.enums import WorkflowNotificationAction +from pyatlan.model.fields.atlan_fields import ( + KeywordField, + NumericField, + RelationField, + TextField, +) + +from .notification import Notification + + +class WorkflowNotification(Notification): + """Description""" + + type_name: str = Field(default="WorkflowNotification", allow_mutation=False) + + @validator("type_name") + def validate_type_name(cls, v): + if v != "WorkflowNotification": + raise ValueError("must be WorkflowNotification") + return v + + def __setattr__(self, name, value): + if name in WorkflowNotification._convenience_properties: + return object.__setattr__(self, name, value) + super().__setattr__(name, value) + + WORKFLOW_NOTIFICATION_WORKFLOW_SLUG: ClassVar[KeywordField] = KeywordField( + "workflowNotificationWorkflowSlug", "workflowNotificationWorkflowSlug" + ) + """ + Slug of the workflow this notification is about. What grouping by workflow groups on, and how a check finds the live notification before deciding whether to raise another. + """ # noqa: E501 + WORKFLOW_NOTIFICATION_LAST_PREFLIGHT_RESULT: ClassVar[TextField] = TextField( + "workflowNotificationLastPreflightResult", + "workflowNotificationLastPreflightResult", + ) + """ + The whole check response as one escaped JSON string, rewritten on every check while the notification is live. Rendered by the widget the interface already has, so nothing is denormalised into attributes. + """ # noqa: E501 + WORKFLOW_NOTIFICATION_PREFLIGHT_FAILURE_COUNT: ClassVar[NumericField] = ( + NumericField( + "workflowNotificationPreflightFailureCount", + "workflowNotificationPreflightFailureCount", + ) + ) + """ + How many times the check has failed for this notification, counted from the failures that raised it and incremented on every failure since. + """ # noqa: E501 + WORKFLOW_NOTIFICATION_LAST_PREFLIGHT_FAILED_AT: ClassVar[NumericField] = ( + NumericField( + "workflowNotificationLastPreflightFailedAt", + "workflowNotificationLastPreflightFailedAt", + ) + ) + """ + When the check last failed. Distinct from when it was last checked, which every write already records. + """ + WORKFLOW_NOTIFICATION_ACTION: ClassVar[KeywordField] = KeywordField( + "workflowNotificationAction", "workflowNotificationAction" + ) + """ + Which option the person took. Declared here rather than on the notification supertype because these values only mean anything for a workflow. + """ # noqa: E501 + + WORKFLOW_NOTIFICATION_WORKFLOW: ClassVar[RelationField] = RelationField( + "workflowNotificationWorkflow" + ) + """ + TBC + """ + + _convenience_properties: ClassVar[List[str]] = [ + "workflow_notification_workflow_slug", + "workflow_notification_last_preflight_result", + "workflow_notification_preflight_failure_count", + "workflow_notification_last_preflight_failed_at", + "workflow_notification_action", + "workflow_notification_workflow", + ] + + @property + def workflow_notification_workflow_slug(self) -> Optional[str]: + return ( + None + if self.attributes is None + else self.attributes.workflow_notification_workflow_slug + ) + + @workflow_notification_workflow_slug.setter + def workflow_notification_workflow_slug( + self, workflow_notification_workflow_slug: Optional[str] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.workflow_notification_workflow_slug = ( + workflow_notification_workflow_slug + ) + + @property + def workflow_notification_last_preflight_result(self) -> Optional[str]: + return ( + None + if self.attributes is None + else self.attributes.workflow_notification_last_preflight_result + ) + + @workflow_notification_last_preflight_result.setter + def workflow_notification_last_preflight_result( + self, workflow_notification_last_preflight_result: Optional[str] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.workflow_notification_last_preflight_result = ( + workflow_notification_last_preflight_result + ) + + @property + def workflow_notification_preflight_failure_count(self) -> Optional[int]: + return ( + None + if self.attributes is None + else self.attributes.workflow_notification_preflight_failure_count + ) + + @workflow_notification_preflight_failure_count.setter + def workflow_notification_preflight_failure_count( + self, workflow_notification_preflight_failure_count: Optional[int] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.workflow_notification_preflight_failure_count = ( + workflow_notification_preflight_failure_count + ) + + @property + def workflow_notification_last_preflight_failed_at(self) -> Optional[datetime]: + return ( + None + if self.attributes is None + else self.attributes.workflow_notification_last_preflight_failed_at + ) + + @workflow_notification_last_preflight_failed_at.setter + def workflow_notification_last_preflight_failed_at( + self, workflow_notification_last_preflight_failed_at: Optional[datetime] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.workflow_notification_last_preflight_failed_at = ( + workflow_notification_last_preflight_failed_at + ) + + @property + def workflow_notification_action(self) -> Optional[WorkflowNotificationAction]: + return ( + None + if self.attributes is None + else self.attributes.workflow_notification_action + ) + + @workflow_notification_action.setter + def workflow_notification_action( + self, workflow_notification_action: Optional[WorkflowNotificationAction] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.workflow_notification_action = workflow_notification_action + + @property + def workflow_notification_workflow(self) -> Optional[AtlanAppWorkflow]: + return ( + None + if self.attributes is None + else self.attributes.workflow_notification_workflow + ) + + @workflow_notification_workflow.setter + def workflow_notification_workflow( + self, workflow_notification_workflow: Optional[AtlanAppWorkflow] + ): + if self.attributes is None: + self.attributes = self.Attributes() + self.attributes.workflow_notification_workflow = workflow_notification_workflow + + class Attributes(Notification.Attributes): + workflow_notification_workflow_slug: Optional[str] = Field( + default=None, description="" + ) + workflow_notification_last_preflight_result: Optional[str] = Field( + default=None, description="" + ) + workflow_notification_preflight_failure_count: Optional[int] = Field( + default=None, description="" + ) + workflow_notification_last_preflight_failed_at: Optional[datetime] = Field( + default=None, description="" + ) + workflow_notification_action: Optional[WorkflowNotificationAction] = Field( + default=None, description="" + ) + workflow_notification_workflow: Optional[AtlanAppWorkflow] = Field( + default=None, description="" + ) # relationship + + attributes: WorkflowNotification.Attributes = Field( + default_factory=lambda: WorkflowNotification.Attributes(), + description=( + "Map of attributes in the instance and their values. " + "The specific keys of this map will vary by type, " + "so are described in the sub-types of this schema." + ), + ) + + +from .atlan_app_workflow import AtlanAppWorkflow # noqa: E402, F401 diff --git a/pyatlan/model/enums.py b/pyatlan/model/enums.py index 9e1905d29..68ec76c7b 100644 --- a/pyatlan/model/enums.py +++ b/pyatlan/model/enums.py @@ -3207,6 +3207,19 @@ class MongoDBCollectionValidationLevel(str, Enum): MODERATE = "MODERATE" +class NotificationState(str, Enum): + GENERATED = "GENERATED" + DELIVERED = "DELIVERED" + ACTIONED = "ACTIONED" + RESOLVED = "RESOLVED" + CLOSED = "CLOSED" + + +class NotificationSurface(str, Enum): + SLACK = "SLACK" + TEAMS = "TEAMS" + + class OpenLineageRunState(str, Enum): START = "START" RUNNING = "RUNNING" @@ -3222,6 +3235,11 @@ class PowerbiEndorsement(str, Enum): MASTER_DATA = "Master Data" +class ProcessLineageDerivation(str, Enum): + STATIC = "STATIC" + RUNTIME = "RUNTIME" + + class QueryUsernameStrategy(str, Enum): CONNECTION_USERNAME = "connectionUsername" ATLAN_USERNAME = "atlanUsername" @@ -3335,6 +3353,15 @@ class TableType(str, Enum): KUDU = "KUDU" +class WorkflowNotificationAction(str, Enum): + EDIT_CREDENTIALS = "EDIT_CREDENTIALS" + RECHECK_NOW = "RECHECK_NOW" + PAUSE_RUNS = "PAUSE_RUNS" + REMOVE_SCHEDULE = "REMOVE_SCHEDULE" + RAISE_TICKET = "RAISE_TICKET" + DISMISS = "DISMISS" + + class WorkflowRunStatus(str, Enum): PENDING = "PENDING" APPROVED = "APPROVED" diff --git a/pyatlan/model/structs.py b/pyatlan/model/structs.py index bb90931a6..87088f6c0 100644 --- a/pyatlan/model/structs.py +++ b/pyatlan/model/structs.py @@ -17,6 +17,7 @@ DataQualityRuleThresholdUnit, FormFieldDimension, FormFieldType, + NotificationSurface, SourceCostUnitType, ) from pyatlan.model.utils import to_camel_case @@ -1000,6 +1001,23 @@ class GoogleLabel(AtlanObject): google_label_value: str = Field(description="") +class NotificationExternalReference(AtlanObject): + """Description""" + + notification_external_reference_surface: Optional[NotificationSurface] = Field( + default=None, description="" + ) + notification_external_reference_ref_url: Optional[str] = Field( + default=None, description="" + ) + notification_external_reference_delivered_at: Optional[datetime] = Field( + default=None, description="" + ) + notification_external_reference_extra_attributes: Optional[Dict[str, str]] = Field( + default=None, description="" + ) + + class SourceTagAttribute(AtlanObject): """Description""" @@ -1118,4 +1136,6 @@ class SourceTagAttribute(AtlanObject): GoogleLabel.update_forward_refs() +NotificationExternalReference.update_forward_refs() + SourceTagAttribute.update_forward_refs() diff --git a/pyatlan_v9/model/assets/__init__.py b/pyatlan_v9/model/assets/__init__.py index 353dc985b..fae0ebd5e 100644 --- a/pyatlan_v9/model/assets/__init__.py +++ b/pyatlan_v9/model/assets/__init__.py @@ -671,6 +671,7 @@ "RelatedNamespace", ], "_init_notebook": ["Notebook", "RelatedNotebook"], + "_init_notification": ["Notification", "RelatedNotification"], "_init_partial": [ "Partial", "PartialField", @@ -1130,6 +1131,10 @@ "UnstructuredObject", ], "_init_workflow": ["RelatedWorkflow", "RelatedWorkflowRun", "Workflow"], + "_init_workflow_notification": [ + "RelatedWorkflowNotification", + "WorkflowNotification", + ], } __getattr__, __dir__, __all__ = lazy.attach( diff --git a/pyatlan_v9/model/assets/__init__.pyi b/pyatlan_v9/model/assets/__init__.pyi index 06a044f88..429a0aee3 100644 --- a/pyatlan_v9/model/assets/__init__.pyi +++ b/pyatlan_v9/model/assets/__init__.pyi @@ -587,6 +587,8 @@ __all__ = [ "RelatedNamespace", "Notebook", "RelatedNotebook", + "Notification", + "RelatedNotification", "Partial", "PartialField", "PartialObject", @@ -992,6 +994,8 @@ __all__ = [ "RelatedWorkflow", "RelatedWorkflowRun", "Workflow", + "RelatedWorkflowNotification", + "WorkflowNotification", ] from .access_control import AccessControl as AccessControl @@ -1675,6 +1679,8 @@ from .namespace_related import RelatedNamespace as RelatedNamespace from .no_sql import NoSQL as NoSQL from .notebook import Notebook as Notebook from .notebook_related import RelatedNotebook as RelatedNotebook +from .notification import Notification as Notification +from .notification_related import RelatedNotification as RelatedNotification from .object_store import ObjectStore as ObjectStore from .partial import Partial as Partial from .partial_field import PartialField as PartialField @@ -2185,5 +2191,9 @@ from .unstructured_related import RelatedUnstructuredFolder as RelatedUnstructur from .unstructured_related import RelatedUnstructuredObject as RelatedUnstructuredObject from .view import View as View from .workflow import Workflow as Workflow +from .workflow_notification import WorkflowNotification as WorkflowNotification +from .workflow_notification_related import ( + RelatedWorkflowNotification as RelatedWorkflowNotification, +) from .workflow_related import RelatedWorkflow as RelatedWorkflow from .workflow_related import RelatedWorkflowRun as RelatedWorkflowRun diff --git a/pyatlan_v9/model/assets/_init_notification.py b/pyatlan_v9/model/assets/_init_notification.py new file mode 100644 index 000000000..53b0a6594 --- /dev/null +++ b/pyatlan_v9/model/assets/_init_notification.py @@ -0,0 +1,17 @@ +# Auto-generated by PythonMsgspecRenderer.pkl - DO NOT EDIT +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2024 Atlan Pte. Ltd. + +""" +Notification module exports. + +This module provides convenient imports for all Notification types and their Related variants. +""" + +from .notification import Notification +from .notification_related import RelatedNotification + +__all__ = [ + "Notification", + "RelatedNotification", +] diff --git a/pyatlan_v9/model/assets/_init_workflow_notification.py b/pyatlan_v9/model/assets/_init_workflow_notification.py new file mode 100644 index 000000000..e0f0d0910 --- /dev/null +++ b/pyatlan_v9/model/assets/_init_workflow_notification.py @@ -0,0 +1,17 @@ +# Auto-generated by PythonMsgspecRenderer.pkl - DO NOT EDIT +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2024 Atlan Pte. Ltd. + +""" +WorkflowNotification module exports. + +This module provides convenient imports for all WorkflowNotification types and their Related variants. +""" + +from .workflow_notification import WorkflowNotification +from .workflow_notification_related import RelatedWorkflowNotification + +__all__ = [ + "RelatedWorkflowNotification", + "WorkflowNotification", +] diff --git a/pyatlan_v9/model/assets/atlan_app_workflow.py b/pyatlan_v9/model/assets/atlan_app_workflow.py index e81eab564..4139e9386 100644 --- a/pyatlan_v9/model/assets/atlan_app_workflow.py +++ b/pyatlan_v9/model/assets/atlan_app_workflow.py @@ -59,6 +59,7 @@ from .schema_registry_related import RelatedSchemaRegistrySubject from .soda_related import RelatedSodaCheck from .spark_related import RelatedSparkJob +from .workflow_notification_related import RelatedWorkflowNotification # ============================================================================= # FLAT ASSET CLASS @@ -123,6 +124,7 @@ class AtlanAppWorkflow(Asset): SODA_CHECKS: ClassVar[Any] = None INPUT_TO_SPARK_JOBS: ClassVar[Any] = None OUTPUT_FROM_SPARK_JOBS: ClassVar[Any] = None + WORKFLOW_NOTIFICATION_NOTIFICATIONS: ClassVar[Any] = None atlan_app_workflow_version: Union[str, None, UnsetType] = UNSET """Version of the workflow.""" @@ -290,6 +292,11 @@ class AtlanAppWorkflow(Asset): output_from_spark_jobs: Union[List[RelatedSparkJob], None, UnsetType] = UNSET """""" + workflow_notification_notifications: Union[ + List[RelatedWorkflowNotification], None, UnsetType + ] = UNSET + """Notifications raised about this workflow.""" + def __post_init__(self) -> None: self.type_name = "AtlanAppWorkflow" @@ -594,6 +601,11 @@ class AtlanAppWorkflowRelationshipAttributes(AssetRelationshipAttributes): output_from_spark_jobs: Union[List[RelatedSparkJob], None, UnsetType] = UNSET """""" + workflow_notification_notifications: Union[ + List[RelatedWorkflowNotification], None, UnsetType + ] = UNSET + """Notifications raised about this workflow.""" + class AtlanAppWorkflowNested(AssetNested): """AtlanAppWorkflow in nested API format for high-performance serialization.""" @@ -652,6 +664,7 @@ class AtlanAppWorkflowNested(AssetNested): "soda_checks", "input_to_spark_jobs", "output_from_spark_jobs", + "workflow_notification_notifications", ] @@ -902,3 +915,6 @@ def _atlan_app_workflow_from_nested_bytes( AtlanAppWorkflow.SODA_CHECKS = RelationField("sodaChecks") AtlanAppWorkflow.INPUT_TO_SPARK_JOBS = RelationField("inputToSparkJobs") AtlanAppWorkflow.OUTPUT_FROM_SPARK_JOBS = RelationField("outputFromSparkJobs") +AtlanAppWorkflow.WORKFLOW_NOTIFICATION_NOTIFICATIONS = RelationField( + "workflowNotificationNotifications" +) diff --git a/pyatlan_v9/model/assets/notification.py b/pyatlan_v9/model/assets/notification.py new file mode 100644 index 000000000..a04932072 --- /dev/null +++ b/pyatlan_v9/model/assets/notification.py @@ -0,0 +1,646 @@ +# Auto-generated by PythonMsgspecRenderer.pkl - DO NOT EDIT +# ruff: noqa: ARG002 +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2024 Atlan Pte. Ltd. + +""" +Notification asset model with flattened inheritance. + +This module provides: +- Notification: Flat asset class (easy to use) +- NotificationAttributes: Nested attributes struct (extends AssetAttributes) +- NotificationNested: Nested API format struct +""" + +from __future__ import annotations + +from typing import Any, ClassVar, Dict, List, Set, Union + +from msgspec import UNSET, UnsetType + +from pyatlan_v9.model.conversion_utils import ( + categorize_relationships, + merge_relationships, +) +from pyatlan_v9.model.serde import Serde, get_serde +from pyatlan_v9.model.transform import register_asset + +from .anomalo_related import RelatedAnomaloCheck +from .app_related import RelatedApplication, RelatedApplicationField +from .asset import ( + _ASSET_REL_FIELDS, + Asset, + AssetAttributes, + AssetNested, + AssetRelationshipAttributes, + _extract_asset_attrs, + _populate_asset_attrs, +) +from .context_related import RelatedContextRepository +from .data_contract_related import RelatedDataContract +from .data_mesh_related import RelatedDataProduct +from .data_quality_related import RelatedDataQualityRule, RelatedMetric +from .gcp_dataplex_related import RelatedGCPDataplexAspectType +from .gtc_related import RelatedAtlasGlossaryTerm +from .monte_carlo_related import RelatedMCIncident, RelatedMCMonitor +from .notification_related import RelatedNotification +from .referenceable_related import RelatedReferenceable +from .resource_related import RelatedFile, RelatedLink, RelatedReadme +from .schema_registry_related import RelatedSchemaRegistrySubject +from .soda_related import RelatedSodaCheck + +# ============================================================================= +# FLAT ASSET CLASS +# ============================================================================= + + +@register_asset +class Notification(Asset): + """ + Base class for a durable, queryable notification record. Never instantiated: only a subtype is ever created, the way nothing creates a bare Catalog. + """ + + NOTIFICATION_STATE: ClassVar[Any] = None + NOTIFICATION_RECIPIENT_USERS: ClassVar[Any] = None + NOTIFICATION_RECIPIENT_GROUPS: ClassVar[Any] = None + NOTIFICATION_RECIPIENT_ROLES: ClassVar[Any] = None + NOTIFICATION_LAST_ACTED_BY: ClassVar[Any] = None + NOTIFICATION_LAST_ACTED_AT: ClassVar[Any] = None + NOTIFICATION_LAST_ACTED_REASON: ClassVar[Any] = None + NOTIFICATION_EXTERNAL_REFERENCES: ClassVar[Any] = None + ANOMALO_CHECKS: ClassVar[Any] = None + APPLICATION: ClassVar[Any] = None + APPLICATION_FIELD: ClassVar[Any] = None + CONTEXT_REPOSITORIES: ClassVar[Any] = None + DATA_CONTRACT_LATEST: ClassVar[Any] = None + DATA_CONTRACT_LATEST_CERTIFIED: ClassVar[Any] = None + OUTPUT_PORT_DATA_PRODUCTS: ClassVar[Any] = None + INPUT_PORT_DATA_PRODUCTS: ClassVar[Any] = None + METRICS: ClassVar[Any] = None + DQ_BASE_DATASET_RULES: ClassVar[Any] = None + DQ_REFERENCE_DATASET_RULES: ClassVar[Any] = None + GCP_DATAPLEX_ASPECT_TYPE_METADATA_ENTITIES: ClassVar[Any] = None + MEANINGS: ClassVar[Any] = None + MC_MONITORS: ClassVar[Any] = None + MC_INCIDENTS: ClassVar[Any] = None + USER_DEF_RELATIONSHIP_TO: ClassVar[Any] = None + USER_DEF_RELATIONSHIP_FROM: ClassVar[Any] = None + FILES: ClassVar[Any] = None + LINKS: ClassVar[Any] = None + README: ClassVar[Any] = None + SCHEMA_REGISTRY_SUBJECTS: ClassVar[Any] = None + SODA_CHECKS: ClassVar[Any] = None + + notification_state: Union[str, None, UnsetType] = UNSET + """Where the notification is in its lifecycle. Also how a check decides whether a live notification already exists before raising another.""" + + notification_recipient_users: Union[Set[str], None, UnsetType] = UNSET + """Usernames the notification was addressed to. Drives the assigned-to-me view together with the group and role recipients.""" + + notification_recipient_groups: Union[Set[str], None, UnsetType] = UNSET + """Identifiers of the Atlan groups the notification is meant for. A group outlives its members, so a rule addressing a group stays generic as people join and leave.""" + + notification_recipient_roles: Union[Set[str], None, UnsetType] = UNSET + """Identifiers of the Atlan roles that need to see the notification. Resolved to individuals by the interface at read time rather than stored.""" + + notification_last_acted_by: Union[str, None, UnsetType] = UNSET + """Username of the person who last moved the state. Declared separately because the write is service-mediated, so the entity's own modifier records the service and not the person.""" + + notification_last_acted_at: Union[int, None, UnsetType] = UNSET + """When a person last moved the state. Read against the creation timestamp this gives time to resolution.""" + + notification_last_acted_reason: Union[str, None, UnsetType] = UNSET + """Reason or note the person gave for the last action. Mandatory for some actions and optional for others, so it is free text rather than a code.""" + + notification_external_references: Union[List[Dict[str, Any]], None, UnsetType] = ( + UNSET + ) + """Where this notification also lives outside Atlan, one entry per surface. This is what lets an action taken in the inbox go back and update the message it came from.""" + + anomalo_checks: Union[List[RelatedAnomaloCheck], None, UnsetType] = UNSET + """Checks that run on this asset.""" + + application: Union[RelatedApplication, None, UnsetType] = UNSET + """Application owning the Asset.""" + + application_field: Union[RelatedApplicationField, None, UnsetType] = UNSET + """ApplicationField owning the Asset.""" + + context_repositories: Union[List[RelatedContextRepository], None, UnsetType] = UNSET + """Context repositories that use this asset as input.""" + + data_contract_latest: Union[RelatedDataContract, None, UnsetType] = UNSET + """Latest version of the data contract (in any status) for this asset.""" + + data_contract_latest_certified: Union[RelatedDataContract, None, UnsetType] = UNSET + """Latest certified version of the data contract for this asset.""" + + output_port_data_products: Union[List[RelatedDataProduct], None, UnsetType] = UNSET + """Data products for which this asset is an output port.""" + + input_port_data_products: Union[List[RelatedDataProduct], None, UnsetType] = UNSET + """Data products for which this asset is an input port.""" + + metrics: Union[List[RelatedMetric], None, UnsetType] = UNSET + """""" + + dq_base_dataset_rules: Union[List[RelatedDataQualityRule], None, UnsetType] = UNSET + """Rules that are applied on this dataset.""" + + dq_reference_dataset_rules: Union[List[RelatedDataQualityRule], None, UnsetType] = ( + UNSET + ) + """Rules where this dataset is referenced.""" + + gcp_dataplex_aspect_type_metadata_entities: Union[ + List[RelatedGCPDataplexAspectType], None, UnsetType + ] = UNSET + """Dataplex entries (assets) that have aspects of this Aspect Type attached.""" + + meanings: Union[List[RelatedAtlasGlossaryTerm], None, UnsetType] = UNSET + """Glossary terms that are linked to this asset.""" + + mc_monitors: Union[List[RelatedMCMonitor], None, UnsetType] = UNSET + """Monitors that observe this asset.""" + + mc_incidents: Union[List[RelatedMCIncident], None, UnsetType] = UNSET + """""" + + user_def_relationship_to: Union[List[RelatedReferenceable], None, UnsetType] = UNSET + """""" + + user_def_relationship_from: Union[List[RelatedReferenceable], None, UnsetType] = ( + UNSET + ) + """""" + + files: Union[List[RelatedFile], None, UnsetType] = UNSET + """""" + + links: Union[List[RelatedLink], None, UnsetType] = UNSET + """Links that are attached to this asset.""" + + readme: Union[RelatedReadme, None, UnsetType] = UNSET + """README that is linked to this asset.""" + + schema_registry_subjects: Union[ + List[RelatedSchemaRegistrySubject], None, UnsetType + ] = UNSET + """Schema registry subjects associated with this asset.""" + + soda_checks: Union[List[RelatedSodaCheck], None, UnsetType] = UNSET + """""" + + def __post_init__(self) -> None: + self.type_name = "Notification" + + # ========================================================================= + # SDK Methods + # ========================================================================= + + def validate(self, for_creation: bool = False) -> None: + """ + Dry-run validation of this Notification instance. + + Checks that required fields (type_name, name, qualified_name) are set. + When ``for_creation=True``, also checks hierarchy-specific fields + (parent references, denormalized attributes) needed to create this asset. + + This is purely opt-in and is NOT called by any serde path — only by + explicit user invocation (e.g., validating JSONL before sending to Atlan). + + Args: + for_creation: If True, also validate fields required for asset creation. + + Raises: + ValueError: If any required fields are missing or invalid. + """ + errors: list[str] = [] + if self.type_name is UNSET: + errors.append("type_name is required") + if self.name is UNSET: + errors.append("name is required") + if self.qualified_name is UNSET or self.qualified_name is None: + errors.append("qualified_name is required") + if errors: + raise ValueError(f"Notification validation failed: {errors}") + + def minimize(self) -> "Notification": + """ + Return a minimal copy of this Notification with only updater-required fields. + + Calls :meth:`validate` first to ensure the instance is valid, then + returns a new Notification with only the fields needed for an update + (qualified_name, name, and any type-specific additional fields). + + Returns: + A new Notification instance with only the minimum required fields. + """ + self.validate() + return Notification(qualified_name=self.qualified_name, name=self.name) + + def relate(self) -> "RelatedNotification": + """ + Create a :class:`RelatedNotification` reference from this instance. + + Returns a lightweight reference suitable for use in relationship + attributes. Prefers ``guid`` if set, otherwise falls back to + ``qualified_name``. + + Returns: + A RelatedNotification reference to this asset. + """ + if self.guid is not UNSET: + return RelatedNotification(guid=self.guid) + return RelatedNotification(qualified_name=self.qualified_name) + + # ========================================================================= + # Optimized Serialization Methods (override Asset base class) + # ========================================================================= + + def to_json(self, nested: bool = True, serde: Serde | None = None) -> str: + """ + Convert to JSON string using optimized nested struct serialization. + + Args: + nested: If True (default), use nested API format. If False, use flat format. + serde: Optional Serde instance for encoder reuse. Uses shared singleton if None. + + Returns: + JSON string representation + """ + if serde is None: + serde = get_serde() + if nested: + return self.to_nested_bytes(serde).decode("utf-8") + else: + return serde.encode(self).decode("utf-8") + + def to_nested_bytes(self, serde: Serde | None = None) -> bytes: + """Serialize to Atlas nested-format JSON bytes (pure msgspec, no dict intermediate).""" + if serde is None: + serde = get_serde() + return _notification_to_nested_bytes(self, serde) + + @staticmethod + def from_json(json_data: str | bytes, serde: Serde | None = None) -> Notification: + """ + Create from JSON string or bytes using optimized nested struct deserialization. + + Args: + json_data: JSON string or bytes to deserialize + serde: Optional Serde instance for decoder reuse. Uses shared singleton if None. + + Returns: + Notification instance + """ + if isinstance(json_data, str): + json_data = json_data.encode("utf-8") + if serde is None: + serde = get_serde() + return _notification_from_nested_bytes(json_data, serde) + + +# ============================================================================= +# NESTED FORMAT CLASSES +# ============================================================================= + + +class NotificationAttributes(AssetAttributes): + """Notification-specific attributes for nested API format.""" + + notification_state: Union[str, None, UnsetType] = UNSET + """Where the notification is in its lifecycle. Also how a check decides whether a live notification already exists before raising another.""" + + notification_recipient_users: Union[Set[str], None, UnsetType] = UNSET + """Usernames the notification was addressed to. Drives the assigned-to-me view together with the group and role recipients.""" + + notification_recipient_groups: Union[Set[str], None, UnsetType] = UNSET + """Identifiers of the Atlan groups the notification is meant for. A group outlives its members, so a rule addressing a group stays generic as people join and leave.""" + + notification_recipient_roles: Union[Set[str], None, UnsetType] = UNSET + """Identifiers of the Atlan roles that need to see the notification. Resolved to individuals by the interface at read time rather than stored.""" + + notification_last_acted_by: Union[str, None, UnsetType] = UNSET + """Username of the person who last moved the state. Declared separately because the write is service-mediated, so the entity's own modifier records the service and not the person.""" + + notification_last_acted_at: Union[int, None, UnsetType] = UNSET + """When a person last moved the state. Read against the creation timestamp this gives time to resolution.""" + + notification_last_acted_reason: Union[str, None, UnsetType] = UNSET + """Reason or note the person gave for the last action. Mandatory for some actions and optional for others, so it is free text rather than a code.""" + + notification_external_references: Union[List[Dict[str, Any]], None, UnsetType] = ( + UNSET + ) + """Where this notification also lives outside Atlan, one entry per surface. This is what lets an action taken in the inbox go back and update the message it came from.""" + + +class NotificationRelationshipAttributes(AssetRelationshipAttributes): + """Notification-specific relationship attributes for nested API format.""" + + anomalo_checks: Union[List[RelatedAnomaloCheck], None, UnsetType] = UNSET + """Checks that run on this asset.""" + + application: Union[RelatedApplication, None, UnsetType] = UNSET + """Application owning the Asset.""" + + application_field: Union[RelatedApplicationField, None, UnsetType] = UNSET + """ApplicationField owning the Asset.""" + + context_repositories: Union[List[RelatedContextRepository], None, UnsetType] = UNSET + """Context repositories that use this asset as input.""" + + data_contract_latest: Union[RelatedDataContract, None, UnsetType] = UNSET + """Latest version of the data contract (in any status) for this asset.""" + + data_contract_latest_certified: Union[RelatedDataContract, None, UnsetType] = UNSET + """Latest certified version of the data contract for this asset.""" + + output_port_data_products: Union[List[RelatedDataProduct], None, UnsetType] = UNSET + """Data products for which this asset is an output port.""" + + input_port_data_products: Union[List[RelatedDataProduct], None, UnsetType] = UNSET + """Data products for which this asset is an input port.""" + + metrics: Union[List[RelatedMetric], None, UnsetType] = UNSET + """""" + + dq_base_dataset_rules: Union[List[RelatedDataQualityRule], None, UnsetType] = UNSET + """Rules that are applied on this dataset.""" + + dq_reference_dataset_rules: Union[List[RelatedDataQualityRule], None, UnsetType] = ( + UNSET + ) + """Rules where this dataset is referenced.""" + + gcp_dataplex_aspect_type_metadata_entities: Union[ + List[RelatedGCPDataplexAspectType], None, UnsetType + ] = UNSET + """Dataplex entries (assets) that have aspects of this Aspect Type attached.""" + + meanings: Union[List[RelatedAtlasGlossaryTerm], None, UnsetType] = UNSET + """Glossary terms that are linked to this asset.""" + + mc_monitors: Union[List[RelatedMCMonitor], None, UnsetType] = UNSET + """Monitors that observe this asset.""" + + mc_incidents: Union[List[RelatedMCIncident], None, UnsetType] = UNSET + """""" + + user_def_relationship_to: Union[List[RelatedReferenceable], None, UnsetType] = UNSET + """""" + + user_def_relationship_from: Union[List[RelatedReferenceable], None, UnsetType] = ( + UNSET + ) + """""" + + files: Union[List[RelatedFile], None, UnsetType] = UNSET + """""" + + links: Union[List[RelatedLink], None, UnsetType] = UNSET + """Links that are attached to this asset.""" + + readme: Union[RelatedReadme, None, UnsetType] = UNSET + """README that is linked to this asset.""" + + schema_registry_subjects: Union[ + List[RelatedSchemaRegistrySubject], None, UnsetType + ] = UNSET + """Schema registry subjects associated with this asset.""" + + soda_checks: Union[List[RelatedSodaCheck], None, UnsetType] = UNSET + """""" + + +class NotificationNested(AssetNested): + """Notification in nested API format for high-performance serialization.""" + + attributes: Union[NotificationAttributes, UnsetType] = UNSET + relationship_attributes: Union[NotificationRelationshipAttributes, UnsetType] = ( + UNSET + ) + append_relationship_attributes: Union[ + NotificationRelationshipAttributes, UnsetType + ] = UNSET + remove_relationship_attributes: Union[ + NotificationRelationshipAttributes, UnsetType + ] = UNSET + + +# ============================================================================= +# CONVERSION HELPERS & CONSTANTS +# ============================================================================= + +_NOTIFICATION_REL_FIELDS: List[str] = [ + *_ASSET_REL_FIELDS, + "anomalo_checks", + "application", + "application_field", + "context_repositories", + "data_contract_latest", + "data_contract_latest_certified", + "output_port_data_products", + "input_port_data_products", + "metrics", + "dq_base_dataset_rules", + "dq_reference_dataset_rules", + "gcp_dataplex_aspect_type_metadata_entities", + "meanings", + "mc_monitors", + "mc_incidents", + "user_def_relationship_to", + "user_def_relationship_from", + "files", + "links", + "readme", + "schema_registry_subjects", + "soda_checks", +] + + +def _populate_notification_attrs( + attrs: NotificationAttributes, obj: Notification +) -> None: + """Populate Notification-specific attributes on the attrs struct.""" + _populate_asset_attrs(attrs, obj) + attrs.notification_state = obj.notification_state + attrs.notification_recipient_users = obj.notification_recipient_users + attrs.notification_recipient_groups = obj.notification_recipient_groups + attrs.notification_recipient_roles = obj.notification_recipient_roles + attrs.notification_last_acted_by = obj.notification_last_acted_by + attrs.notification_last_acted_at = obj.notification_last_acted_at + attrs.notification_last_acted_reason = obj.notification_last_acted_reason + attrs.notification_external_references = obj.notification_external_references + + +def _extract_notification_attrs(attrs: NotificationAttributes) -> dict: + """Extract all Notification attributes from the attrs struct into a flat dict.""" + result = _extract_asset_attrs(attrs) + result["notification_state"] = attrs.notification_state + result["notification_recipient_users"] = attrs.notification_recipient_users + result["notification_recipient_groups"] = attrs.notification_recipient_groups + result["notification_recipient_roles"] = attrs.notification_recipient_roles + result["notification_last_acted_by"] = attrs.notification_last_acted_by + result["notification_last_acted_at"] = attrs.notification_last_acted_at + result["notification_last_acted_reason"] = attrs.notification_last_acted_reason + result["notification_external_references"] = attrs.notification_external_references + return result + + +# ============================================================================= +# CONVERSION FUNCTIONS +# ============================================================================= + + +def _notification_to_nested(notification: Notification) -> NotificationNested: + """Convert flat Notification to nested format.""" + attrs = NotificationAttributes() + _populate_notification_attrs(attrs, notification) + # Categorize relationships by save semantic (REPLACE, APPEND, REMOVE) + replace_rels, append_rels, remove_rels = categorize_relationships( + notification, _NOTIFICATION_REL_FIELDS, NotificationRelationshipAttributes + ) + return NotificationNested( + guid=notification.guid, + type_name=notification.type_name, + status=notification.status, + version=notification.version, + create_time=notification.create_time, + update_time=notification.update_time, + created_by=notification.created_by, + updated_by=notification.updated_by, + classifications=notification.classifications, + classification_names=notification.classification_names, + meanings=notification.meanings, + labels=notification.labels, + business_attributes=notification.business_attributes, + custom_attributes=notification.custom_attributes, + pending_tasks=notification.pending_tasks, + proxy=notification.proxy, + is_incomplete=notification.is_incomplete, + provenance_type=notification.provenance_type, + home_id=notification.home_id, + depth=notification.depth, + immediate_upstream=notification.immediate_upstream, + immediate_downstream=notification.immediate_downstream, + attributes=attrs, + relationship_attributes=replace_rels, + append_relationship_attributes=append_rels, + remove_relationship_attributes=remove_rels, + ) + + +def _notification_from_nested(nested: NotificationNested) -> Notification: + """Convert nested format to flat Notification.""" + attrs = ( + nested.attributes + if nested.attributes is not UNSET + else NotificationAttributes() + ) + # Merge relationships from all three buckets + merged_rels = merge_relationships( + nested.relationship_attributes, + nested.append_relationship_attributes, + nested.remove_relationship_attributes, + _NOTIFICATION_REL_FIELDS, + NotificationRelationshipAttributes, + ) + return Notification( + guid=nested.guid, + type_name=nested.type_name, + status=nested.status, + version=nested.version, + create_time=nested.create_time, + update_time=nested.update_time, + created_by=nested.created_by, + updated_by=nested.updated_by, + classifications=nested.classifications, + classification_names=nested.classification_names, + labels=nested.labels, + business_attributes=nested.business_attributes, + custom_attributes=nested.custom_attributes, + pending_tasks=nested.pending_tasks, + proxy=nested.proxy, + is_incomplete=nested.is_incomplete, + provenance_type=nested.provenance_type, + home_id=nested.home_id, + depth=nested.depth, + immediate_upstream=nested.immediate_upstream, + immediate_downstream=nested.immediate_downstream, + **_extract_notification_attrs(attrs), + # Merged relationship attributes + **merged_rels, + ) + + +def _notification_to_nested_bytes(notification: Notification, serde: Serde) -> bytes: + """Convert flat Notification to nested JSON bytes.""" + return serde.encode(_notification_to_nested(notification)) + + +def _notification_from_nested_bytes(data: bytes, serde: Serde) -> Notification: + """Convert nested JSON bytes to flat Notification.""" + nested = serde.decode(data, NotificationNested) + return _notification_from_nested(nested) + + +# --------------------------------------------------------------------------- +# Deferred field descriptor initialization +# --------------------------------------------------------------------------- +from pyatlan.model.fields.atlan_fields import ( # noqa: E402 + KeywordField, + NumericField, + RelationField, + TextField, +) + +Notification.NOTIFICATION_STATE = KeywordField("notificationState", "notificationState") +Notification.NOTIFICATION_RECIPIENT_USERS = KeywordField( + "notificationRecipientUsers", "notificationRecipientUsers" +) +Notification.NOTIFICATION_RECIPIENT_GROUPS = KeywordField( + "notificationRecipientGroups", "notificationRecipientGroups" +) +Notification.NOTIFICATION_RECIPIENT_ROLES = KeywordField( + "notificationRecipientRoles", "notificationRecipientRoles" +) +Notification.NOTIFICATION_LAST_ACTED_BY = KeywordField( + "notificationLastActedBy", "notificationLastActedBy" +) +Notification.NOTIFICATION_LAST_ACTED_AT = NumericField( + "notificationLastActedAt", "notificationLastActedAt" +) +Notification.NOTIFICATION_LAST_ACTED_REASON = TextField( + "notificationLastActedReason", "notificationLastActedReason" +) +Notification.NOTIFICATION_EXTERNAL_REFERENCES = KeywordField( + "notificationExternalReferences", "notificationExternalReferences" +) +Notification.ANOMALO_CHECKS = RelationField("anomaloChecks") +Notification.APPLICATION = RelationField("application") +Notification.APPLICATION_FIELD = RelationField("applicationField") +Notification.CONTEXT_REPOSITORIES = RelationField("contextRepositories") +Notification.DATA_CONTRACT_LATEST = RelationField("dataContractLatest") +Notification.DATA_CONTRACT_LATEST_CERTIFIED = RelationField( + "dataContractLatestCertified" +) +Notification.OUTPUT_PORT_DATA_PRODUCTS = RelationField("outputPortDataProducts") +Notification.INPUT_PORT_DATA_PRODUCTS = RelationField("inputPortDataProducts") +Notification.METRICS = RelationField("metrics") +Notification.DQ_BASE_DATASET_RULES = RelationField("dqBaseDatasetRules") +Notification.DQ_REFERENCE_DATASET_RULES = RelationField("dqReferenceDatasetRules") +Notification.GCP_DATAPLEX_ASPECT_TYPE_METADATA_ENTITIES = RelationField( + "gcpDataplexAspectTypeMetadataEntities" +) +Notification.MEANINGS = RelationField("meanings") +Notification.MC_MONITORS = RelationField("mcMonitors") +Notification.MC_INCIDENTS = RelationField("mcIncidents") +Notification.USER_DEF_RELATIONSHIP_TO = RelationField("userDefRelationshipTo") +Notification.USER_DEF_RELATIONSHIP_FROM = RelationField("userDefRelationshipFrom") +Notification.FILES = RelationField("files") +Notification.LINKS = RelationField("links") +Notification.README = RelationField("readme") +Notification.SCHEMA_REGISTRY_SUBJECTS = RelationField("schemaRegistrySubjects") +Notification.SODA_CHECKS = RelationField("sodaChecks") diff --git a/pyatlan_v9/model/assets/notification_related.py b/pyatlan_v9/model/assets/notification_related.py new file mode 100644 index 000000000..3291ef2d5 --- /dev/null +++ b/pyatlan_v9/model/assets/notification_related.py @@ -0,0 +1,64 @@ +# Auto-generated by PythonMsgspecRenderer.pkl - DO NOT EDIT +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2024 Atlan Pte. Ltd. + +""" +Related type classes for Notification module. + +This module contains all Related{Type} classes for the Notification type hierarchy. +These classes are used for relationship attributes to reference related entities. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Set, Union + +from msgspec import UNSET, UnsetType + +from .asset_related import RelatedAsset +from .referenceable_related import RelatedReferenceable + +__all__ = [ + "RelatedNotification", +] + + +class RelatedNotification(RelatedAsset): + """ + Related entity reference for Notification assets. + + Extends RelatedAsset with Notification-specific attributes. + """ + + # type_name inherited from parent with default=UNSET + # __post_init__ sets it to "Notification" so it serializes correctly + + notification_state: Union[str, None, UnsetType] = UNSET + """Where the notification is in its lifecycle. Also how a check decides whether a live notification already exists before raising another.""" + + notification_recipient_users: Union[Set[str], None, UnsetType] = UNSET + """Usernames the notification was addressed to. Drives the assigned-to-me view together with the group and role recipients.""" + + notification_recipient_groups: Union[Set[str], None, UnsetType] = UNSET + """Identifiers of the Atlan groups the notification is meant for. A group outlives its members, so a rule addressing a group stays generic as people join and leave.""" + + notification_recipient_roles: Union[Set[str], None, UnsetType] = UNSET + """Identifiers of the Atlan roles that need to see the notification. Resolved to individuals by the interface at read time rather than stored.""" + + notification_last_acted_by: Union[str, None, UnsetType] = UNSET + """Username of the person who last moved the state. Declared separately because the write is service-mediated, so the entity's own modifier records the service and not the person.""" + + notification_last_acted_at: Union[int, None, UnsetType] = UNSET + """When a person last moved the state. Read against the creation timestamp this gives time to resolution.""" + + notification_last_acted_reason: Union[str, None, UnsetType] = UNSET + """Reason or note the person gave for the last action. Mandatory for some actions and optional for others, so it is free text rather than a code.""" + + notification_external_references: Union[List[Dict[str, Any]], None, UnsetType] = ( + UNSET + ) + """Where this notification also lives outside Atlan, one entry per surface. This is what lets an action taken in the inbox go back and update the message it came from.""" + + def __post_init__(self) -> None: + RelatedReferenceable.__post_init__(self) + self.type_name = "Notification" diff --git a/pyatlan_v9/model/assets/workflow_notification.py b/pyatlan_v9/model/assets/workflow_notification.py new file mode 100644 index 000000000..9e3238f1d --- /dev/null +++ b/pyatlan_v9/model/assets/workflow_notification.py @@ -0,0 +1,767 @@ +# Auto-generated by PythonMsgspecRenderer.pkl - DO NOT EDIT +# ruff: noqa: ARG002 +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2024 Atlan Pte. Ltd. + +""" +WorkflowNotification asset model with flattened inheritance. + +This module provides: +- WorkflowNotification: Flat asset class (easy to use) +- WorkflowNotificationAttributes: Nested attributes struct (extends AssetAttributes) +- WorkflowNotificationNested: Nested API format struct +""" + +from __future__ import annotations + +import re +from typing import Any, ClassVar, Dict, List, Set, Union + +from msgspec import UNSET, UnsetType + +from pyatlan_v9.model.conversion_utils import ( + categorize_relationships, + merge_relationships, +) +from pyatlan_v9.model.serde import Serde, get_serde +from pyatlan_v9.model.transform import register_asset + +from .anomalo_related import RelatedAnomaloCheck +from .app_related import RelatedApplication, RelatedApplicationField +from .asset import ( + _ASSET_REL_FIELDS, + Asset, + AssetAttributes, + AssetNested, + AssetRelationshipAttributes, + _extract_asset_attrs, + _populate_asset_attrs, +) +from .atlan_app_related import RelatedAtlanAppWorkflow +from .context_related import RelatedContextRepository +from .data_contract_related import RelatedDataContract +from .data_mesh_related import RelatedDataProduct +from .data_quality_related import RelatedDataQualityRule, RelatedMetric +from .gcp_dataplex_related import RelatedGCPDataplexAspectType +from .gtc_related import RelatedAtlasGlossaryTerm +from .monte_carlo_related import RelatedMCIncident, RelatedMCMonitor +from .referenceable_related import RelatedReferenceable +from .resource_related import RelatedFile, RelatedLink, RelatedReadme +from .schema_registry_related import RelatedSchemaRegistrySubject +from .soda_related import RelatedSodaCheck +from .workflow_notification_related import RelatedWorkflowNotification + +# ============================================================================= +# FLAT ASSET CLASS +# ============================================================================= + + +@register_asset +class WorkflowNotification(Asset): + """ + A notification about an Atlan app workflow. The only notification type created today, and the one every future workflow-related notification should use. + """ + + WORKFLOW_NOTIFICATION_WORKFLOW_SLUG: ClassVar[Any] = None + WORKFLOW_NOTIFICATION_LAST_PREFLIGHT_RESULT: ClassVar[Any] = None + WORKFLOW_NOTIFICATION_PREFLIGHT_FAILURE_COUNT: ClassVar[Any] = None + WORKFLOW_NOTIFICATION_LAST_PREFLIGHT_FAILED_AT: ClassVar[Any] = None + WORKFLOW_NOTIFICATION_ACTION: ClassVar[Any] = None + NOTIFICATION_STATE: ClassVar[Any] = None + NOTIFICATION_RECIPIENT_USERS: ClassVar[Any] = None + NOTIFICATION_RECIPIENT_GROUPS: ClassVar[Any] = None + NOTIFICATION_RECIPIENT_ROLES: ClassVar[Any] = None + NOTIFICATION_LAST_ACTED_BY: ClassVar[Any] = None + NOTIFICATION_LAST_ACTED_AT: ClassVar[Any] = None + NOTIFICATION_LAST_ACTED_REASON: ClassVar[Any] = None + NOTIFICATION_EXTERNAL_REFERENCES: ClassVar[Any] = None + ANOMALO_CHECKS: ClassVar[Any] = None + APPLICATION: ClassVar[Any] = None + APPLICATION_FIELD: ClassVar[Any] = None + CONTEXT_REPOSITORIES: ClassVar[Any] = None + DATA_CONTRACT_LATEST: ClassVar[Any] = None + DATA_CONTRACT_LATEST_CERTIFIED: ClassVar[Any] = None + OUTPUT_PORT_DATA_PRODUCTS: ClassVar[Any] = None + INPUT_PORT_DATA_PRODUCTS: ClassVar[Any] = None + METRICS: ClassVar[Any] = None + DQ_BASE_DATASET_RULES: ClassVar[Any] = None + DQ_REFERENCE_DATASET_RULES: ClassVar[Any] = None + GCP_DATAPLEX_ASPECT_TYPE_METADATA_ENTITIES: ClassVar[Any] = None + MEANINGS: ClassVar[Any] = None + MC_MONITORS: ClassVar[Any] = None + MC_INCIDENTS: ClassVar[Any] = None + USER_DEF_RELATIONSHIP_TO: ClassVar[Any] = None + USER_DEF_RELATIONSHIP_FROM: ClassVar[Any] = None + FILES: ClassVar[Any] = None + LINKS: ClassVar[Any] = None + README: ClassVar[Any] = None + SCHEMA_REGISTRY_SUBJECTS: ClassVar[Any] = None + SODA_CHECKS: ClassVar[Any] = None + WORKFLOW_NOTIFICATION_WORKFLOW: ClassVar[Any] = None + + workflow_notification_workflow_slug: Union[str, None, UnsetType] = UNSET + """Slug of the workflow this notification is about. What grouping by workflow groups on, and how a check finds the live notification before deciding whether to raise another.""" + + workflow_notification_last_preflight_result: Union[str, None, UnsetType] = UNSET + """The whole check response as one escaped JSON string, rewritten on every check while the notification is live. Rendered by the widget the interface already has, so nothing is denormalised into attributes.""" + + workflow_notification_preflight_failure_count: Union[int, None, UnsetType] = UNSET + """How many times the check has failed for this notification, counted from the failures that raised it and incremented on every failure since.""" + + workflow_notification_last_preflight_failed_at: Union[int, None, UnsetType] = UNSET + """When the check last failed. Distinct from when it was last checked, which every write already records.""" + + workflow_notification_action: Union[str, None, UnsetType] = UNSET + """Which option the person took. Declared here rather than on the notification supertype because these values only mean anything for a workflow.""" + + notification_state: Union[str, None, UnsetType] = UNSET + """Where the notification is in its lifecycle. Also how a check decides whether a live notification already exists before raising another.""" + + notification_recipient_users: Union[Set[str], None, UnsetType] = UNSET + """Usernames the notification was addressed to. Drives the assigned-to-me view together with the group and role recipients.""" + + notification_recipient_groups: Union[Set[str], None, UnsetType] = UNSET + """Identifiers of the Atlan groups the notification is meant for. A group outlives its members, so a rule addressing a group stays generic as people join and leave.""" + + notification_recipient_roles: Union[Set[str], None, UnsetType] = UNSET + """Identifiers of the Atlan roles that need to see the notification. Resolved to individuals by the interface at read time rather than stored.""" + + notification_last_acted_by: Union[str, None, UnsetType] = UNSET + """Username of the person who last moved the state. Declared separately because the write is service-mediated, so the entity's own modifier records the service and not the person.""" + + notification_last_acted_at: Union[int, None, UnsetType] = UNSET + """When a person last moved the state. Read against the creation timestamp this gives time to resolution.""" + + notification_last_acted_reason: Union[str, None, UnsetType] = UNSET + """Reason or note the person gave for the last action. Mandatory for some actions and optional for others, so it is free text rather than a code.""" + + notification_external_references: Union[List[Dict[str, Any]], None, UnsetType] = ( + UNSET + ) + """Where this notification also lives outside Atlan, one entry per surface. This is what lets an action taken in the inbox go back and update the message it came from.""" + + anomalo_checks: Union[List[RelatedAnomaloCheck], None, UnsetType] = UNSET + """Checks that run on this asset.""" + + application: Union[RelatedApplication, None, UnsetType] = UNSET + """Application owning the Asset.""" + + application_field: Union[RelatedApplicationField, None, UnsetType] = UNSET + """ApplicationField owning the Asset.""" + + context_repositories: Union[List[RelatedContextRepository], None, UnsetType] = UNSET + """Context repositories that use this asset as input.""" + + data_contract_latest: Union[RelatedDataContract, None, UnsetType] = UNSET + """Latest version of the data contract (in any status) for this asset.""" + + data_contract_latest_certified: Union[RelatedDataContract, None, UnsetType] = UNSET + """Latest certified version of the data contract for this asset.""" + + output_port_data_products: Union[List[RelatedDataProduct], None, UnsetType] = UNSET + """Data products for which this asset is an output port.""" + + input_port_data_products: Union[List[RelatedDataProduct], None, UnsetType] = UNSET + """Data products for which this asset is an input port.""" + + metrics: Union[List[RelatedMetric], None, UnsetType] = UNSET + """""" + + dq_base_dataset_rules: Union[List[RelatedDataQualityRule], None, UnsetType] = UNSET + """Rules that are applied on this dataset.""" + + dq_reference_dataset_rules: Union[List[RelatedDataQualityRule], None, UnsetType] = ( + UNSET + ) + """Rules where this dataset is referenced.""" + + gcp_dataplex_aspect_type_metadata_entities: Union[ + List[RelatedGCPDataplexAspectType], None, UnsetType + ] = UNSET + """Dataplex entries (assets) that have aspects of this Aspect Type attached.""" + + meanings: Union[List[RelatedAtlasGlossaryTerm], None, UnsetType] = UNSET + """Glossary terms that are linked to this asset.""" + + mc_monitors: Union[List[RelatedMCMonitor], None, UnsetType] = UNSET + """Monitors that observe this asset.""" + + mc_incidents: Union[List[RelatedMCIncident], None, UnsetType] = UNSET + """""" + + user_def_relationship_to: Union[List[RelatedReferenceable], None, UnsetType] = UNSET + """""" + + user_def_relationship_from: Union[List[RelatedReferenceable], None, UnsetType] = ( + UNSET + ) + """""" + + files: Union[List[RelatedFile], None, UnsetType] = UNSET + """""" + + links: Union[List[RelatedLink], None, UnsetType] = UNSET + """Links that are attached to this asset.""" + + readme: Union[RelatedReadme, None, UnsetType] = UNSET + """README that is linked to this asset.""" + + schema_registry_subjects: Union[ + List[RelatedSchemaRegistrySubject], None, UnsetType + ] = UNSET + """Schema registry subjects associated with this asset.""" + + soda_checks: Union[List[RelatedSodaCheck], None, UnsetType] = UNSET + """""" + + workflow_notification_workflow: Union[RelatedAtlanAppWorkflow, None, UnsetType] = ( + UNSET + ) + """The workflow this notification is about.""" + + def __post_init__(self) -> None: + self.type_name = "WorkflowNotification" + + # ========================================================================= + # SDK Methods + # ========================================================================= + + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/[^/]+/[^/]+$") + + def validate(self, for_creation: bool = False) -> None: + """ + Dry-run validation of this WorkflowNotification instance. + + Checks that required fields (type_name, name, qualified_name) are set. + When ``for_creation=True``, also checks hierarchy-specific fields + (parent references, denormalized attributes) needed to create this asset. + + This is purely opt-in and is NOT called by any serde path — only by + explicit user invocation (e.g., validating JSONL before sending to Atlan). + + Args: + for_creation: If True, also validate fields required for asset creation. + + Raises: + ValueError: If any required fields are missing or invalid. + """ + errors: list[str] = [] + if self.type_name is UNSET: + errors.append("type_name is required") + if self.name is UNSET: + errors.append("name is required") + if self.qualified_name is UNSET or self.qualified_name is None: + errors.append("qualified_name is required") + elif not self._QUALIFIED_NAME_PATTERN.match(self.qualified_name): + errors.append( + f"qualified_name '{self.qualified_name}' does not match expected " + f"pattern: {self._QUALIFIED_NAME_PATTERN.pattern}" + ) + if for_creation: + if self.connection_qualified_name is UNSET: + errors.append("connection_qualified_name is required for creation") + if errors: + raise ValueError(f"WorkflowNotification validation failed: {errors}") + + def minimize(self) -> "WorkflowNotification": + """ + Return a minimal copy of this WorkflowNotification with only updater-required fields. + + Calls :meth:`validate` first to ensure the instance is valid, then + returns a new WorkflowNotification with only the fields needed for an update + (qualified_name, name, and any type-specific additional fields). + + Returns: + A new WorkflowNotification instance with only the minimum required fields. + """ + self.validate() + return WorkflowNotification(qualified_name=self.qualified_name, name=self.name) + + def relate(self) -> "RelatedWorkflowNotification": + """ + Create a :class:`RelatedWorkflowNotification` reference from this instance. + + Returns a lightweight reference suitable for use in relationship + attributes. Prefers ``guid`` if set, otherwise falls back to + ``qualified_name``. + + Returns: + A RelatedWorkflowNotification reference to this asset. + """ + if self.guid is not UNSET: + return RelatedWorkflowNotification(guid=self.guid) + return RelatedWorkflowNotification(qualified_name=self.qualified_name) + + # ========================================================================= + # Optimized Serialization Methods (override Asset base class) + # ========================================================================= + + def to_json(self, nested: bool = True, serde: Serde | None = None) -> str: + """ + Convert to JSON string using optimized nested struct serialization. + + Args: + nested: If True (default), use nested API format. If False, use flat format. + serde: Optional Serde instance for encoder reuse. Uses shared singleton if None. + + Returns: + JSON string representation + """ + if serde is None: + serde = get_serde() + if nested: + return self.to_nested_bytes(serde).decode("utf-8") + else: + return serde.encode(self).decode("utf-8") + + def to_nested_bytes(self, serde: Serde | None = None) -> bytes: + """Serialize to Atlas nested-format JSON bytes (pure msgspec, no dict intermediate).""" + if serde is None: + serde = get_serde() + return _workflow_notification_to_nested_bytes(self, serde) + + @staticmethod + def from_json( + json_data: str | bytes, serde: Serde | None = None + ) -> WorkflowNotification: + """ + Create from JSON string or bytes using optimized nested struct deserialization. + + Args: + json_data: JSON string or bytes to deserialize + serde: Optional Serde instance for decoder reuse. Uses shared singleton if None. + + Returns: + WorkflowNotification instance + """ + if isinstance(json_data, str): + json_data = json_data.encode("utf-8") + if serde is None: + serde = get_serde() + return _workflow_notification_from_nested_bytes(json_data, serde) + + +# ============================================================================= +# NESTED FORMAT CLASSES +# ============================================================================= + + +class WorkflowNotificationAttributes(AssetAttributes): + """WorkflowNotification-specific attributes for nested API format.""" + + workflow_notification_workflow_slug: Union[str, None, UnsetType] = UNSET + """Slug of the workflow this notification is about. What grouping by workflow groups on, and how a check finds the live notification before deciding whether to raise another.""" + + workflow_notification_last_preflight_result: Union[str, None, UnsetType] = UNSET + """The whole check response as one escaped JSON string, rewritten on every check while the notification is live. Rendered by the widget the interface already has, so nothing is denormalised into attributes.""" + + workflow_notification_preflight_failure_count: Union[int, None, UnsetType] = UNSET + """How many times the check has failed for this notification, counted from the failures that raised it and incremented on every failure since.""" + + workflow_notification_last_preflight_failed_at: Union[int, None, UnsetType] = UNSET + """When the check last failed. Distinct from when it was last checked, which every write already records.""" + + workflow_notification_action: Union[str, None, UnsetType] = UNSET + """Which option the person took. Declared here rather than on the notification supertype because these values only mean anything for a workflow.""" + + notification_state: Union[str, None, UnsetType] = UNSET + """Where the notification is in its lifecycle. Also how a check decides whether a live notification already exists before raising another.""" + + notification_recipient_users: Union[Set[str], None, UnsetType] = UNSET + """Usernames the notification was addressed to. Drives the assigned-to-me view together with the group and role recipients.""" + + notification_recipient_groups: Union[Set[str], None, UnsetType] = UNSET + """Identifiers of the Atlan groups the notification is meant for. A group outlives its members, so a rule addressing a group stays generic as people join and leave.""" + + notification_recipient_roles: Union[Set[str], None, UnsetType] = UNSET + """Identifiers of the Atlan roles that need to see the notification. Resolved to individuals by the interface at read time rather than stored.""" + + notification_last_acted_by: Union[str, None, UnsetType] = UNSET + """Username of the person who last moved the state. Declared separately because the write is service-mediated, so the entity's own modifier records the service and not the person.""" + + notification_last_acted_at: Union[int, None, UnsetType] = UNSET + """When a person last moved the state. Read against the creation timestamp this gives time to resolution.""" + + notification_last_acted_reason: Union[str, None, UnsetType] = UNSET + """Reason or note the person gave for the last action. Mandatory for some actions and optional for others, so it is free text rather than a code.""" + + notification_external_references: Union[List[Dict[str, Any]], None, UnsetType] = ( + UNSET + ) + """Where this notification also lives outside Atlan, one entry per surface. This is what lets an action taken in the inbox go back and update the message it came from.""" + + +class WorkflowNotificationRelationshipAttributes(AssetRelationshipAttributes): + """WorkflowNotification-specific relationship attributes for nested API format.""" + + anomalo_checks: Union[List[RelatedAnomaloCheck], None, UnsetType] = UNSET + """Checks that run on this asset.""" + + application: Union[RelatedApplication, None, UnsetType] = UNSET + """Application owning the Asset.""" + + application_field: Union[RelatedApplicationField, None, UnsetType] = UNSET + """ApplicationField owning the Asset.""" + + context_repositories: Union[List[RelatedContextRepository], None, UnsetType] = UNSET + """Context repositories that use this asset as input.""" + + data_contract_latest: Union[RelatedDataContract, None, UnsetType] = UNSET + """Latest version of the data contract (in any status) for this asset.""" + + data_contract_latest_certified: Union[RelatedDataContract, None, UnsetType] = UNSET + """Latest certified version of the data contract for this asset.""" + + output_port_data_products: Union[List[RelatedDataProduct], None, UnsetType] = UNSET + """Data products for which this asset is an output port.""" + + input_port_data_products: Union[List[RelatedDataProduct], None, UnsetType] = UNSET + """Data products for which this asset is an input port.""" + + metrics: Union[List[RelatedMetric], None, UnsetType] = UNSET + """""" + + dq_base_dataset_rules: Union[List[RelatedDataQualityRule], None, UnsetType] = UNSET + """Rules that are applied on this dataset.""" + + dq_reference_dataset_rules: Union[List[RelatedDataQualityRule], None, UnsetType] = ( + UNSET + ) + """Rules where this dataset is referenced.""" + + gcp_dataplex_aspect_type_metadata_entities: Union[ + List[RelatedGCPDataplexAspectType], None, UnsetType + ] = UNSET + """Dataplex entries (assets) that have aspects of this Aspect Type attached.""" + + meanings: Union[List[RelatedAtlasGlossaryTerm], None, UnsetType] = UNSET + """Glossary terms that are linked to this asset.""" + + mc_monitors: Union[List[RelatedMCMonitor], None, UnsetType] = UNSET + """Monitors that observe this asset.""" + + mc_incidents: Union[List[RelatedMCIncident], None, UnsetType] = UNSET + """""" + + user_def_relationship_to: Union[List[RelatedReferenceable], None, UnsetType] = UNSET + """""" + + user_def_relationship_from: Union[List[RelatedReferenceable], None, UnsetType] = ( + UNSET + ) + """""" + + files: Union[List[RelatedFile], None, UnsetType] = UNSET + """""" + + links: Union[List[RelatedLink], None, UnsetType] = UNSET + """Links that are attached to this asset.""" + + readme: Union[RelatedReadme, None, UnsetType] = UNSET + """README that is linked to this asset.""" + + schema_registry_subjects: Union[ + List[RelatedSchemaRegistrySubject], None, UnsetType + ] = UNSET + """Schema registry subjects associated with this asset.""" + + soda_checks: Union[List[RelatedSodaCheck], None, UnsetType] = UNSET + """""" + + workflow_notification_workflow: Union[RelatedAtlanAppWorkflow, None, UnsetType] = ( + UNSET + ) + """The workflow this notification is about.""" + + +class WorkflowNotificationNested(AssetNested): + """WorkflowNotification in nested API format for high-performance serialization.""" + + attributes: Union[WorkflowNotificationAttributes, UnsetType] = UNSET + relationship_attributes: Union[ + WorkflowNotificationRelationshipAttributes, UnsetType + ] = UNSET + append_relationship_attributes: Union[ + WorkflowNotificationRelationshipAttributes, UnsetType + ] = UNSET + remove_relationship_attributes: Union[ + WorkflowNotificationRelationshipAttributes, UnsetType + ] = UNSET + + +# ============================================================================= +# CONVERSION HELPERS & CONSTANTS +# ============================================================================= + +_WORKFLOW_NOTIFICATION_REL_FIELDS: List[str] = [ + *_ASSET_REL_FIELDS, + "anomalo_checks", + "application", + "application_field", + "context_repositories", + "data_contract_latest", + "data_contract_latest_certified", + "output_port_data_products", + "input_port_data_products", + "metrics", + "dq_base_dataset_rules", + "dq_reference_dataset_rules", + "gcp_dataplex_aspect_type_metadata_entities", + "meanings", + "mc_monitors", + "mc_incidents", + "user_def_relationship_to", + "user_def_relationship_from", + "files", + "links", + "readme", + "schema_registry_subjects", + "soda_checks", + "workflow_notification_workflow", +] + + +def _populate_workflow_notification_attrs( + attrs: WorkflowNotificationAttributes, obj: WorkflowNotification +) -> None: + """Populate WorkflowNotification-specific attributes on the attrs struct.""" + _populate_asset_attrs(attrs, obj) + attrs.workflow_notification_workflow_slug = obj.workflow_notification_workflow_slug + attrs.workflow_notification_last_preflight_result = ( + obj.workflow_notification_last_preflight_result + ) + attrs.workflow_notification_preflight_failure_count = ( + obj.workflow_notification_preflight_failure_count + ) + attrs.workflow_notification_last_preflight_failed_at = ( + obj.workflow_notification_last_preflight_failed_at + ) + attrs.workflow_notification_action = obj.workflow_notification_action + attrs.notification_state = obj.notification_state + attrs.notification_recipient_users = obj.notification_recipient_users + attrs.notification_recipient_groups = obj.notification_recipient_groups + attrs.notification_recipient_roles = obj.notification_recipient_roles + attrs.notification_last_acted_by = obj.notification_last_acted_by + attrs.notification_last_acted_at = obj.notification_last_acted_at + attrs.notification_last_acted_reason = obj.notification_last_acted_reason + attrs.notification_external_references = obj.notification_external_references + + +def _extract_workflow_notification_attrs(attrs: WorkflowNotificationAttributes) -> dict: + """Extract all WorkflowNotification attributes from the attrs struct into a flat dict.""" + result = _extract_asset_attrs(attrs) + result["workflow_notification_workflow_slug"] = ( + attrs.workflow_notification_workflow_slug + ) + result["workflow_notification_last_preflight_result"] = ( + attrs.workflow_notification_last_preflight_result + ) + result["workflow_notification_preflight_failure_count"] = ( + attrs.workflow_notification_preflight_failure_count + ) + result["workflow_notification_last_preflight_failed_at"] = ( + attrs.workflow_notification_last_preflight_failed_at + ) + result["workflow_notification_action"] = attrs.workflow_notification_action + result["notification_state"] = attrs.notification_state + result["notification_recipient_users"] = attrs.notification_recipient_users + result["notification_recipient_groups"] = attrs.notification_recipient_groups + result["notification_recipient_roles"] = attrs.notification_recipient_roles + result["notification_last_acted_by"] = attrs.notification_last_acted_by + result["notification_last_acted_at"] = attrs.notification_last_acted_at + result["notification_last_acted_reason"] = attrs.notification_last_acted_reason + result["notification_external_references"] = attrs.notification_external_references + return result + + +# ============================================================================= +# CONVERSION FUNCTIONS +# ============================================================================= + + +def _workflow_notification_to_nested( + workflow_notification: WorkflowNotification, +) -> WorkflowNotificationNested: + """Convert flat WorkflowNotification to nested format.""" + attrs = WorkflowNotificationAttributes() + _populate_workflow_notification_attrs(attrs, workflow_notification) + # Categorize relationships by save semantic (REPLACE, APPEND, REMOVE) + replace_rels, append_rels, remove_rels = categorize_relationships( + workflow_notification, + _WORKFLOW_NOTIFICATION_REL_FIELDS, + WorkflowNotificationRelationshipAttributes, + ) + return WorkflowNotificationNested( + guid=workflow_notification.guid, + type_name=workflow_notification.type_name, + status=workflow_notification.status, + version=workflow_notification.version, + create_time=workflow_notification.create_time, + update_time=workflow_notification.update_time, + created_by=workflow_notification.created_by, + updated_by=workflow_notification.updated_by, + classifications=workflow_notification.classifications, + classification_names=workflow_notification.classification_names, + meanings=workflow_notification.meanings, + labels=workflow_notification.labels, + business_attributes=workflow_notification.business_attributes, + custom_attributes=workflow_notification.custom_attributes, + pending_tasks=workflow_notification.pending_tasks, + proxy=workflow_notification.proxy, + is_incomplete=workflow_notification.is_incomplete, + provenance_type=workflow_notification.provenance_type, + home_id=workflow_notification.home_id, + depth=workflow_notification.depth, + immediate_upstream=workflow_notification.immediate_upstream, + immediate_downstream=workflow_notification.immediate_downstream, + attributes=attrs, + relationship_attributes=replace_rels, + append_relationship_attributes=append_rels, + remove_relationship_attributes=remove_rels, + ) + + +def _workflow_notification_from_nested( + nested: WorkflowNotificationNested, +) -> WorkflowNotification: + """Convert nested format to flat WorkflowNotification.""" + attrs = ( + nested.attributes + if nested.attributes is not UNSET + else WorkflowNotificationAttributes() + ) + # Merge relationships from all three buckets + merged_rels = merge_relationships( + nested.relationship_attributes, + nested.append_relationship_attributes, + nested.remove_relationship_attributes, + _WORKFLOW_NOTIFICATION_REL_FIELDS, + WorkflowNotificationRelationshipAttributes, + ) + return WorkflowNotification( + guid=nested.guid, + type_name=nested.type_name, + status=nested.status, + version=nested.version, + create_time=nested.create_time, + update_time=nested.update_time, + created_by=nested.created_by, + updated_by=nested.updated_by, + classifications=nested.classifications, + classification_names=nested.classification_names, + labels=nested.labels, + business_attributes=nested.business_attributes, + custom_attributes=nested.custom_attributes, + pending_tasks=nested.pending_tasks, + proxy=nested.proxy, + is_incomplete=nested.is_incomplete, + provenance_type=nested.provenance_type, + home_id=nested.home_id, + depth=nested.depth, + immediate_upstream=nested.immediate_upstream, + immediate_downstream=nested.immediate_downstream, + **_extract_workflow_notification_attrs(attrs), + # Merged relationship attributes + **merged_rels, + ) + + +def _workflow_notification_to_nested_bytes( + workflow_notification: WorkflowNotification, serde: Serde +) -> bytes: + """Convert flat WorkflowNotification to nested JSON bytes.""" + return serde.encode(_workflow_notification_to_nested(workflow_notification)) + + +def _workflow_notification_from_nested_bytes( + data: bytes, serde: Serde +) -> WorkflowNotification: + """Convert nested JSON bytes to flat WorkflowNotification.""" + nested = serde.decode(data, WorkflowNotificationNested) + return _workflow_notification_from_nested(nested) + + +# --------------------------------------------------------------------------- +# Deferred field descriptor initialization +# --------------------------------------------------------------------------- +from pyatlan.model.fields.atlan_fields import ( # noqa: E402 + KeywordField, + NumericField, + RelationField, + TextField, +) + +WorkflowNotification.WORKFLOW_NOTIFICATION_WORKFLOW_SLUG = KeywordField( + "workflowNotificationWorkflowSlug", "workflowNotificationWorkflowSlug" +) +WorkflowNotification.WORKFLOW_NOTIFICATION_LAST_PREFLIGHT_RESULT = TextField( + "workflowNotificationLastPreflightResult", "workflowNotificationLastPreflightResult" +) +WorkflowNotification.WORKFLOW_NOTIFICATION_PREFLIGHT_FAILURE_COUNT = NumericField( + "workflowNotificationPreflightFailureCount", + "workflowNotificationPreflightFailureCount", +) +WorkflowNotification.WORKFLOW_NOTIFICATION_LAST_PREFLIGHT_FAILED_AT = NumericField( + "workflowNotificationLastPreflightFailedAt", + "workflowNotificationLastPreflightFailedAt", +) +WorkflowNotification.WORKFLOW_NOTIFICATION_ACTION = KeywordField( + "workflowNotificationAction", "workflowNotificationAction" +) +WorkflowNotification.NOTIFICATION_STATE = KeywordField( + "notificationState", "notificationState" +) +WorkflowNotification.NOTIFICATION_RECIPIENT_USERS = KeywordField( + "notificationRecipientUsers", "notificationRecipientUsers" +) +WorkflowNotification.NOTIFICATION_RECIPIENT_GROUPS = KeywordField( + "notificationRecipientGroups", "notificationRecipientGroups" +) +WorkflowNotification.NOTIFICATION_RECIPIENT_ROLES = KeywordField( + "notificationRecipientRoles", "notificationRecipientRoles" +) +WorkflowNotification.NOTIFICATION_LAST_ACTED_BY = KeywordField( + "notificationLastActedBy", "notificationLastActedBy" +) +WorkflowNotification.NOTIFICATION_LAST_ACTED_AT = NumericField( + "notificationLastActedAt", "notificationLastActedAt" +) +WorkflowNotification.NOTIFICATION_LAST_ACTED_REASON = TextField( + "notificationLastActedReason", "notificationLastActedReason" +) +WorkflowNotification.NOTIFICATION_EXTERNAL_REFERENCES = KeywordField( + "notificationExternalReferences", "notificationExternalReferences" +) +WorkflowNotification.ANOMALO_CHECKS = RelationField("anomaloChecks") +WorkflowNotification.APPLICATION = RelationField("application") +WorkflowNotification.APPLICATION_FIELD = RelationField("applicationField") +WorkflowNotification.CONTEXT_REPOSITORIES = RelationField("contextRepositories") +WorkflowNotification.DATA_CONTRACT_LATEST = RelationField("dataContractLatest") +WorkflowNotification.DATA_CONTRACT_LATEST_CERTIFIED = RelationField( + "dataContractLatestCertified" +) +WorkflowNotification.OUTPUT_PORT_DATA_PRODUCTS = RelationField("outputPortDataProducts") +WorkflowNotification.INPUT_PORT_DATA_PRODUCTS = RelationField("inputPortDataProducts") +WorkflowNotification.METRICS = RelationField("metrics") +WorkflowNotification.DQ_BASE_DATASET_RULES = RelationField("dqBaseDatasetRules") +WorkflowNotification.DQ_REFERENCE_DATASET_RULES = RelationField( + "dqReferenceDatasetRules" +) +WorkflowNotification.GCP_DATAPLEX_ASPECT_TYPE_METADATA_ENTITIES = RelationField( + "gcpDataplexAspectTypeMetadataEntities" +) +WorkflowNotification.MEANINGS = RelationField("meanings") +WorkflowNotification.MC_MONITORS = RelationField("mcMonitors") +WorkflowNotification.MC_INCIDENTS = RelationField("mcIncidents") +WorkflowNotification.USER_DEF_RELATIONSHIP_TO = RelationField("userDefRelationshipTo") +WorkflowNotification.USER_DEF_RELATIONSHIP_FROM = RelationField( + "userDefRelationshipFrom" +) +WorkflowNotification.FILES = RelationField("files") +WorkflowNotification.LINKS = RelationField("links") +WorkflowNotification.README = RelationField("readme") +WorkflowNotification.SCHEMA_REGISTRY_SUBJECTS = RelationField("schemaRegistrySubjects") +WorkflowNotification.SODA_CHECKS = RelationField("sodaChecks") +WorkflowNotification.WORKFLOW_NOTIFICATION_WORKFLOW = RelationField( + "workflowNotificationWorkflow" +) diff --git a/pyatlan_v9/model/assets/workflow_notification_related.py b/pyatlan_v9/model/assets/workflow_notification_related.py new file mode 100644 index 000000000..25a0b8f35 --- /dev/null +++ b/pyatlan_v9/model/assets/workflow_notification_related.py @@ -0,0 +1,53 @@ +# Auto-generated by PythonMsgspecRenderer.pkl - DO NOT EDIT +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2024 Atlan Pte. Ltd. + +""" +Related type classes for WorkflowNotification module. + +This module contains all Related{Type} classes for the WorkflowNotification type hierarchy. +These classes are used for relationship attributes to reference related entities. +""" + +from __future__ import annotations + +from typing import Union + +from msgspec import UNSET, UnsetType + +from .notification_related import RelatedNotification +from .referenceable_related import RelatedReferenceable + +__all__ = [ + "RelatedWorkflowNotification", +] + + +class RelatedWorkflowNotification(RelatedNotification): + """ + Related entity reference for WorkflowNotification assets. + + Extends RelatedNotification with WorkflowNotification-specific attributes. + """ + + # type_name inherited from parent with default=UNSET + # __post_init__ sets it to "WorkflowNotification" so it serializes correctly + + workflow_notification_workflow_slug: Union[str, None, UnsetType] = UNSET + """Slug of the workflow this notification is about. What grouping by workflow groups on, and how a check finds the live notification before deciding whether to raise another.""" + + workflow_notification_last_preflight_result: Union[str, None, UnsetType] = UNSET + """The whole check response as one escaped JSON string, rewritten on every check while the notification is live. Rendered by the widget the interface already has, so nothing is denormalised into attributes.""" + + workflow_notification_preflight_failure_count: Union[int, None, UnsetType] = UNSET + """How many times the check has failed for this notification, counted from the failures that raised it and incremented on every failure since.""" + + workflow_notification_last_preflight_failed_at: Union[int, None, UnsetType] = UNSET + """When the check last failed. Distinct from when it was last checked, which every write already records.""" + + workflow_notification_action: Union[str, None, UnsetType] = UNSET + """Which option the person took. Declared here rather than on the notification supertype because these values only mean anything for a workflow.""" + + def __post_init__(self) -> None: + RelatedReferenceable.__post_init__(self) + self.type_name = "WorkflowNotification"