diff --git a/pyatlan/generator/templates/methods/asset/quick_sight_analysis.jinja2 b/pyatlan/generator/templates/methods/asset/quick_sight_analysis.jinja2 new file mode 100644 index 000000000..6ca84b566 --- /dev/null +++ b/pyatlan/generator/templates/methods/asset/quick_sight_analysis.jinja2 @@ -0,0 +1,43 @@ + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + ) -> QuickSightAnalysis: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_analysis_folders: List[str], + ) -> QuickSightAnalysis: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_analysis_folders: Optional[List[str]] = None, + ) -> QuickSightAnalysis: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + attributes = QuickSightAnalysis.Attributes.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + quick_sight_analysis_folders=quick_sight_analysis_folders, + ) + return cls(attributes=attributes) diff --git a/pyatlan/generator/templates/methods/asset/quick_sight_analysis_visual.jinja2 b/pyatlan/generator/templates/methods/asset/quick_sight_analysis_visual.jinja2 new file mode 100644 index 000000000..b617cd86e --- /dev/null +++ b/pyatlan/generator/templates/methods/asset/quick_sight_analysis_visual.jinja2 @@ -0,0 +1,63 @@ + + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_analysis_qualified_name: str, + ) -> QuickSightAnalysisVisual: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_analysis_qualified_name: str, + connection_qualified_name: str, + ) -> QuickSightAnalysisVisual: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_analysis_qualified_name: str, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightAnalysisVisual: + validate_required_fields( + [ + "name", + "quick_sight_id", + "quick_sight_sheet_name", + "quick_sight_sheet_id", + "quick_sight_analysis_qualified_name", + ], + [ + name, + quick_sight_id, + quick_sight_sheet_name, + quick_sight_sheet_id, + quick_sight_analysis_qualified_name, + ], + ) + attributes = QuickSightAnalysisVisual.Attributes.creator( + name=name, + quick_sight_id=quick_sight_id, + quick_sight_sheet_id=quick_sight_sheet_id, + quick_sight_sheet_name=quick_sight_sheet_name, + quick_sight_analysis_qualified_name=quick_sight_analysis_qualified_name, + connection_qualified_name=connection_qualified_name, + ) + return cls(attributes=attributes) diff --git a/pyatlan/generator/templates/methods/asset/quick_sight_dashboard.jinja2 b/pyatlan/generator/templates/methods/asset/quick_sight_dashboard.jinja2 new file mode 100644 index 000000000..ff130189c --- /dev/null +++ b/pyatlan/generator/templates/methods/asset/quick_sight_dashboard.jinja2 @@ -0,0 +1,43 @@ + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + ) -> QuickSightDashboard: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dashboard_folders: List[str], + ) -> QuickSightDashboard: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dashboard_folders: Optional[List[str]] = None, + ) -> QuickSightDashboard: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + attributes = QuickSightDashboard.Attributes.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + quick_sight_dashboard_folders=quick_sight_dashboard_folders, + ) + return cls(attributes=attributes) diff --git a/pyatlan/generator/templates/methods/asset/quick_sight_dashboard_visual.jinja2 b/pyatlan/generator/templates/methods/asset/quick_sight_dashboard_visual.jinja2 new file mode 100644 index 000000000..1b9a62c56 --- /dev/null +++ b/pyatlan/generator/templates/methods/asset/quick_sight_dashboard_visual.jinja2 @@ -0,0 +1,63 @@ + + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_dashboard_qualified_name: str, + ) -> QuickSightDashboardVisual: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_dashboard_qualified_name: str, + connection_qualified_name: str, + ) -> QuickSightDashboardVisual: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_dashboard_qualified_name: str, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightDashboardVisual: + validate_required_fields( + [ + "name", + "quick_sight_id", + "quick_sight_sheet_id", + "quick_sight_sheet_name", + "quick_sight_dashboard_qualified_name", + ], + [ + name, + quick_sight_id, + quick_sight_sheet_id, + quick_sight_sheet_name, + quick_sight_dashboard_qualified_name, + ], + ) + attributes = QuickSightDashboardVisual.Attributes.creator( + name=name, + quick_sight_id=quick_sight_id, + quick_sight_sheet_id=quick_sight_sheet_id, + quick_sight_sheet_name=quick_sight_sheet_name, + quick_sight_dashboard_qualified_name=quick_sight_dashboard_qualified_name, + connection_qualified_name=connection_qualified_name, + ) + return cls(attributes=attributes) diff --git a/pyatlan/generator/templates/methods/asset/quick_sight_dataset.jinja2 b/pyatlan/generator/templates/methods/asset/quick_sight_dataset.jinja2 new file mode 100644 index 000000000..2d8ef1c89 --- /dev/null +++ b/pyatlan/generator/templates/methods/asset/quick_sight_dataset.jinja2 @@ -0,0 +1,46 @@ + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + ) -> QuickSightDataset: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dataset_import_mode: QuickSightDatasetImportMode, + quick_sight_dataset_folders: List[str], + ) -> QuickSightDataset: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dataset_import_mode: Optional[QuickSightDatasetImportMode] = None, + quick_sight_dataset_folders: Optional[List[str]] = None, + ) -> QuickSightDataset: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + attributes = QuickSightDataset.Attributes.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + quick_sight_dataset_import_mode=quick_sight_dataset_import_mode, + quick_sight_dataset_folders=quick_sight_dataset_folders, + ) + return cls(attributes=attributes) diff --git a/pyatlan/generator/templates/methods/asset/quick_sight_dataset_field.jinja2 b/pyatlan/generator/templates/methods/asset/quick_sight_dataset_field.jinja2 new file mode 100644 index 000000000..716ae731a --- /dev/null +++ b/pyatlan/generator/templates/methods/asset/quick_sight_dataset_field.jinja2 @@ -0,0 +1,46 @@ + + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_dataset_qualified_name: str, + quick_sight_id: str, + ) -> QuickSightDatasetField: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_dataset_qualified_name: str, + quick_sight_id: str, + quick_sight_dataset_field_type: QuickSightDatasetFieldType, + connection_qualified_name: str, + ) -> QuickSightDatasetField: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_dataset_qualified_name: str, + quick_sight_dataset_field_type: Optional[QuickSightDatasetFieldType] = None, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightDatasetField: + validate_required_fields( + ["name", "quick_sight_id", "quick_sight_dataset_qualified_name"], + [name, quick_sight_id, quick_sight_dataset_qualified_name], + ) + attributes = QuickSightDatasetField.Attributes.creator( + name=name, + quick_sight_id=quick_sight_id, + connection_qualified_name=connection_qualified_name, + quick_sight_dataset_qualified_name=quick_sight_dataset_qualified_name, + quick_sight_dataset_field_type=quick_sight_dataset_field_type, + ) + return cls(attributes=attributes) diff --git a/pyatlan/generator/templates/methods/asset/quick_sight_folder.jinja2 b/pyatlan/generator/templates/methods/asset/quick_sight_folder.jinja2 new file mode 100644 index 000000000..6095a298b --- /dev/null +++ b/pyatlan/generator/templates/methods/asset/quick_sight_folder.jinja2 @@ -0,0 +1,43 @@ + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + ) -> QuickSightFolder: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_folder_type: QuickSightFolderType, + ) -> QuickSightFolder: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_folder_type: Optional[QuickSightFolderType] = None, + ) -> QuickSightFolder: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + attributes = QuickSightFolder.Attributes.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + quick_sight_folder_type=quick_sight_folder_type, + ) + return cls(attributes=attributes) diff --git a/pyatlan/generator/templates/methods/attribute/quick_sight_analysis.jinja2 b/pyatlan/generator/templates/methods/attribute/quick_sight_analysis.jinja2 new file mode 100644 index 000000000..e54551a9e --- /dev/null +++ b/pyatlan/generator/templates/methods/attribute/quick_sight_analysis.jinja2 @@ -0,0 +1,32 @@ + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_analysis_folders: Optional[List[str]] = None, + ) -> QuickSightAnalysis.Attributes: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + folders = None + if quick_sight_analysis_folders: + folders = [ + QuickSightFolder.ref_by_qualified_name(quick_sight_folder_qn) + for quick_sight_folder_qn in quick_sight_analysis_folders + ] + + return QuickSightAnalysis.Attributes( + name=name, + quick_sight_id=quick_sight_id, + qualified_name=f"{connection_qualified_name}/{quick_sight_id}", + connection_qualified_name=connection_qualified_name, + connector_name=AtlanConnectorType.get_connector_name( + connection_qualified_name + ), + quick_sight_analysis_folders=folders, + ) diff --git a/pyatlan/generator/templates/methods/attribute/quick_sight_analysis_visual.jinja2 b/pyatlan/generator/templates/methods/attribute/quick_sight_analysis_visual.jinja2 new file mode 100644 index 000000000..1c5970732 --- /dev/null +++ b/pyatlan/generator/templates/methods/attribute/quick_sight_analysis_visual.jinja2 @@ -0,0 +1,51 @@ + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_analysis_qualified_name: str, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightAnalysisVisual.Attributes: + validate_required_fields( + [ + "name", + "quick_sight_id", + "quick_sight_sheet_id", + "quick_sight_sheet_name", + "quick_sight_analysis_qualified_name", + ], + [ + name, + quick_sight_id, + quick_sight_sheet_id, + quick_sight_sheet_name, + quick_sight_analysis_qualified_name, + ], + ) + assert quick_sight_analysis_qualified_name + if connection_qualified_name: + connector_name = AtlanConnectorType.get_connector_name( + connection_qualified_name + ) + else: + connection_qn, connector_name = AtlanConnectorType.get_connector_name( + quick_sight_analysis_qualified_name, + "quick_sight_analysis_qualified_name", + 4, + ) + connection_qualified_name = connection_qualified_name or connection_qn + return QuickSightAnalysisVisual.Attributes( + name=name, + qualified_name=f"{quick_sight_analysis_qualified_name}/{quick_sight_sheet_id}/{quick_sight_id}", + quick_sight_id=quick_sight_id, + quick_sight_sheet_id=quick_sight_sheet_id, + quick_sight_sheet_name=quick_sight_sheet_name, + quick_sight_analysis_qualified_name=quick_sight_analysis_qualified_name, + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) diff --git a/pyatlan/generator/templates/methods/attribute/quick_sight_dashboard.jinja2 b/pyatlan/generator/templates/methods/attribute/quick_sight_dashboard.jinja2 new file mode 100644 index 000000000..f53dbf6b0 --- /dev/null +++ b/pyatlan/generator/templates/methods/attribute/quick_sight_dashboard.jinja2 @@ -0,0 +1,32 @@ + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dashboard_folders: Optional[List[str]] = None, + ) -> QuickSightDashboard.Attributes: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + folders = None + if quick_sight_dashboard_folders: + folders = [ + QuickSightFolder.ref_by_qualified_name(quick_sight_folder_qn) + for quick_sight_folder_qn in quick_sight_dashboard_folders + ] + + return QuickSightDashboard.Attributes( + name=name, + quick_sight_id=quick_sight_id, + qualified_name=f"{connection_qualified_name}/{quick_sight_id}", + connection_qualified_name=connection_qualified_name, + connector_name=AtlanConnectorType.get_connector_name( + connection_qualified_name + ), + quick_sight_dashboard_folders=folders, + ) diff --git a/pyatlan/generator/templates/methods/attribute/quick_sight_dashboard_visual.jinja2 b/pyatlan/generator/templates/methods/attribute/quick_sight_dashboard_visual.jinja2 new file mode 100644 index 000000000..1a872568f --- /dev/null +++ b/pyatlan/generator/templates/methods/attribute/quick_sight_dashboard_visual.jinja2 @@ -0,0 +1,54 @@ + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_dashboard_qualified_name: str, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightDashboardVisual.Attributes: + validate_required_fields( + [ + "name", + "quick_sight_id", + "quick_sight_sheet_name", + "quick_sight_sheet_id", + "quick_sight_dashboard_qualified_name", + ], + [ + name, + quick_sight_id, + quick_sight_sheet_name, + quick_sight_sheet_id, + quick_sight_dashboard_qualified_name, + ], + ) + assert quick_sight_dashboard_qualified_name + if connection_qualified_name: + connector_name = AtlanConnectorType.get_connector_name( + connection_qualified_name + ) + else: + connection_qn, connector_name = AtlanConnectorType.get_connector_name( + quick_sight_dashboard_qualified_name, + "quick_sight_dashboard_qualified_name", + 4, + ) + connection_qualified_name = connection_qualified_name or connection_qn + return QuickSightDashboardVisual.Attributes( + name=name, + qualified_name=f"{quick_sight_dashboard_qualified_name}/{quick_sight_sheet_id}/{quick_sight_id}", + quick_sight_id=quick_sight_id, + quick_sight_sheet_id=quick_sight_sheet_id, + quick_sight_sheet_name=quick_sight_sheet_name, + quick_sight_dashboard_qualified_name=quick_sight_dashboard_qualified_name, + quick_sight_dashboard=QuickSightDashboard.ref_by_qualified_name( + quick_sight_dashboard_qualified_name + ), + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) diff --git a/pyatlan/generator/templates/methods/attribute/quick_sight_dataset.jinja2 b/pyatlan/generator/templates/methods/attribute/quick_sight_dataset.jinja2 new file mode 100644 index 000000000..e9fef959e --- /dev/null +++ b/pyatlan/generator/templates/methods/attribute/quick_sight_dataset.jinja2 @@ -0,0 +1,36 @@ + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dataset_import_mode: Optional[ + QuickSightDatasetImportMode + ] = None, + quick_sight_dataset_folders: Optional[List[str]] = None, + ) -> QuickSightDataset.Attributes: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + folders = None + if quick_sight_dataset_folders: + folders = [ + QuickSightFolder.ref_by_qualified_name(quick_sight_folder_qn) + for quick_sight_folder_qn in quick_sight_dataset_folders + ] + + return QuickSightDataset.Attributes( + name=name, + quick_sight_id=quick_sight_id, + qualified_name=f"{connection_qualified_name}/{quick_sight_id}", + connection_qualified_name=connection_qualified_name, + connector_name=AtlanConnectorType.get_connector_name( + connection_qualified_name + ), + quick_sight_dataset_import_mode=quick_sight_dataset_import_mode, + quick_sight_dataset_folders=folders, + ) diff --git a/pyatlan/generator/templates/methods/attribute/quick_sight_dataset_field.jinja2 b/pyatlan/generator/templates/methods/attribute/quick_sight_dataset_field.jinja2 new file mode 100644 index 000000000..b203b9d98 --- /dev/null +++ b/pyatlan/generator/templates/methods/attribute/quick_sight_dataset_field.jinja2 @@ -0,0 +1,40 @@ + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_dataset_qualified_name: str, + quick_sight_id: str, + quick_sight_dataset_field_type: Optional[QuickSightDatasetFieldType] = None, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightDatasetField.Attributes: + validate_required_fields( + ["name", "quick_sight_id", "quick_sight_dataset_qualified_name"], + [name, quick_sight_id, quick_sight_dataset_qualified_name], + ) + assert quick_sight_dataset_qualified_name + if connection_qualified_name: + connector_name = AtlanConnectorType.get_connector_name( + connection_qualified_name + ) + else: + connection_qn, connector_name = AtlanConnectorType.get_connector_name( + quick_sight_dataset_qualified_name, + "quick_sight_dataset_qualified_name", + 4, + ) + connection_qualified_name = connection_qualified_name or connection_qn + return QuickSightDatasetField.Attributes( + name=name, + quick_sight_dataset_qualified_name=quick_sight_dataset_qualified_name, + quick_sight_id=quick_sight_id, + quick_sight_dataset=QuickSightDataset.ref_by_qualified_name( + quick_sight_dataset_qualified_name + ), + qualified_name=f"{quick_sight_dataset_qualified_name}/{quick_sight_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + quick_sight_dataset_field_type=quick_sight_dataset_field_type, + ) diff --git a/pyatlan/generator/templates/methods/attribute/quick_sight_folder.jinja2 b/pyatlan/generator/templates/methods/attribute/quick_sight_folder.jinja2 new file mode 100644 index 000000000..17ff1999c --- /dev/null +++ b/pyatlan/generator/templates/methods/attribute/quick_sight_folder.jinja2 @@ -0,0 +1,25 @@ + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_folder_type: Optional[QuickSightFolderType] = None, + ) -> QuickSightFolder.Attributes: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + return QuickSightFolder.Attributes( + name=name, + quick_sight_id=quick_sight_id, + qualified_name=f"{connection_qualified_name}/{quick_sight_id}", + connection_qualified_name=connection_qualified_name, + connector_name=AtlanConnectorType.get_connector_name( + connection_qualified_name + ), + quick_sight_folder_type=quick_sight_folder_type, + ) diff --git a/pyatlan/model/assets/quick_sight_analysis.py b/pyatlan/model/assets/quick_sight_analysis.py index 53b9de313..2d430a72f 100644 --- a/pyatlan/model/assets/quick_sight_analysis.py +++ b/pyatlan/model/assets/quick_sight_analysis.py @@ -4,12 +4,13 @@ from __future__ import annotations -from typing import ClassVar, List, Optional, Set +from typing import ClassVar, List, Optional, Set, overload from pydantic.v1 import Field, validator -from pyatlan.model.enums import QuickSightAnalysisStatus +from pyatlan.model.enums import AtlanConnectorType, QuickSightAnalysisStatus from pyatlan.model.fields.atlan_fields import KeywordField, RelationField, TextField +from pyatlan.utils import init_guid, validate_required_fields from .quick_sight import QuickSight @@ -17,6 +18,49 @@ class QuickSightAnalysis(QuickSight): """Description""" + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + ) -> QuickSightAnalysis: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_analysis_folders: List[str], + ) -> QuickSightAnalysis: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_analysis_folders: Optional[List[str]] = None, + ) -> QuickSightAnalysis: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + attributes = QuickSightAnalysis.Attributes.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + quick_sight_analysis_folders=quick_sight_analysis_folders, + ) + return cls(attributes=attributes) + type_name: str = Field(default="QuickSightAnalysis", allow_mutation=False) @validator("type_name") @@ -200,6 +244,38 @@ class Attributes(QuickSight.Attributes): default=None, description="" ) # relationship + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_analysis_folders: Optional[List[str]] = None, + ) -> QuickSightAnalysis.Attributes: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + folders = None + if quick_sight_analysis_folders: + folders = [ + QuickSightFolder.ref_by_qualified_name(quick_sight_folder_qn) + for quick_sight_folder_qn in quick_sight_analysis_folders + ] + + return QuickSightAnalysis.Attributes( + name=name, + quick_sight_id=quick_sight_id, + qualified_name=f"{connection_qualified_name}/{quick_sight_id}", + connection_qualified_name=connection_qualified_name, + connector_name=AtlanConnectorType.get_connector_name( + connection_qualified_name + ), + quick_sight_analysis_folders=folders, + ) + attributes: QuickSightAnalysis.Attributes = Field( default_factory=lambda: QuickSightAnalysis.Attributes(), description=( diff --git a/pyatlan/model/assets/quick_sight_analysis_visual.py b/pyatlan/model/assets/quick_sight_analysis_visual.py index 34581dd05..60a05a982 100644 --- a/pyatlan/model/assets/quick_sight_analysis_visual.py +++ b/pyatlan/model/assets/quick_sight_analysis_visual.py @@ -4,11 +4,13 @@ from __future__ import annotations -from typing import ClassVar, List, Optional +from typing import ClassVar, List, Optional, overload from pydantic.v1 import Field, validator +from pyatlan.model.enums import AtlanConnectorType from pyatlan.model.fields.atlan_fields import KeywordTextField, RelationField +from pyatlan.utils import init_guid, validate_required_fields from .quick_sight import QuickSight @@ -16,6 +18,69 @@ class QuickSightAnalysisVisual(QuickSight): """Description""" + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_analysis_qualified_name: str, + ) -> QuickSightAnalysisVisual: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_analysis_qualified_name: str, + connection_qualified_name: str, + ) -> QuickSightAnalysisVisual: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_analysis_qualified_name: str, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightAnalysisVisual: + validate_required_fields( + [ + "name", + "quick_sight_id", + "quick_sight_sheet_name", + "quick_sight_sheet_id", + "quick_sight_analysis_qualified_name", + ], + [ + name, + quick_sight_id, + quick_sight_sheet_name, + quick_sight_sheet_id, + quick_sight_analysis_qualified_name, + ], + ) + attributes = QuickSightAnalysisVisual.Attributes.creator( + name=name, + quick_sight_id=quick_sight_id, + quick_sight_sheet_id=quick_sight_sheet_id, + quick_sight_sheet_name=quick_sight_sheet_name, + quick_sight_analysis_qualified_name=quick_sight_analysis_qualified_name, + connection_qualified_name=connection_qualified_name, + ) + return cls(attributes=attributes) + type_name: str = Field(default="QuickSightAnalysisVisual", allow_mutation=False) @validator("type_name") @@ -84,6 +149,57 @@ class Attributes(QuickSight.Attributes): default=None, description="" ) # relationship + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_analysis_qualified_name: str, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightAnalysisVisual.Attributes: + validate_required_fields( + [ + "name", + "quick_sight_id", + "quick_sight_sheet_id", + "quick_sight_sheet_name", + "quick_sight_analysis_qualified_name", + ], + [ + name, + quick_sight_id, + quick_sight_sheet_id, + quick_sight_sheet_name, + quick_sight_analysis_qualified_name, + ], + ) + assert quick_sight_analysis_qualified_name + if connection_qualified_name: + connector_name = AtlanConnectorType.get_connector_name( + connection_qualified_name + ) + else: + connection_qn, connector_name = AtlanConnectorType.get_connector_name( + quick_sight_analysis_qualified_name, + "quick_sight_analysis_qualified_name", + 4, + ) + connection_qualified_name = connection_qualified_name or connection_qn + return QuickSightAnalysisVisual.Attributes( + name=name, + qualified_name=f"{quick_sight_analysis_qualified_name}/{quick_sight_sheet_id}/{quick_sight_id}", + quick_sight_id=quick_sight_id, + quick_sight_sheet_id=quick_sight_sheet_id, + quick_sight_sheet_name=quick_sight_sheet_name, + quick_sight_analysis_qualified_name=quick_sight_analysis_qualified_name, + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) + attributes: QuickSightAnalysisVisual.Attributes = Field( default_factory=lambda: QuickSightAnalysisVisual.Attributes(), description=( diff --git a/pyatlan/model/assets/quick_sight_dashboard.py b/pyatlan/model/assets/quick_sight_dashboard.py index 905900419..018c87481 100644 --- a/pyatlan/model/assets/quick_sight_dashboard.py +++ b/pyatlan/model/assets/quick_sight_dashboard.py @@ -5,11 +5,13 @@ from __future__ import annotations from datetime import datetime -from typing import ClassVar, List, Optional +from typing import ClassVar, List, Optional, overload from pydantic.v1 import Field, validator +from pyatlan.model.enums import AtlanConnectorType from pyatlan.model.fields.atlan_fields import NumericField, RelationField +from pyatlan.utils import init_guid, validate_required_fields from .quick_sight import QuickSight @@ -17,6 +19,49 @@ class QuickSightDashboard(QuickSight): """Description""" + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + ) -> QuickSightDashboard: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dashboard_folders: List[str], + ) -> QuickSightDashboard: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dashboard_folders: Optional[List[str]] = None, + ) -> QuickSightDashboard: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + attributes = QuickSightDashboard.Attributes.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + quick_sight_dashboard_folders=quick_sight_dashboard_folders, + ) + return cls(attributes=attributes) + type_name: str = Field(default="QuickSightDashboard", allow_mutation=False) @validator("type_name") @@ -150,6 +195,38 @@ class Attributes(QuickSight.Attributes): Field(default=None, description="") ) # relationship + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dashboard_folders: Optional[List[str]] = None, + ) -> QuickSightDashboard.Attributes: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + folders = None + if quick_sight_dashboard_folders: + folders = [ + QuickSightFolder.ref_by_qualified_name(quick_sight_folder_qn) + for quick_sight_folder_qn in quick_sight_dashboard_folders + ] + + return QuickSightDashboard.Attributes( + name=name, + quick_sight_id=quick_sight_id, + qualified_name=f"{connection_qualified_name}/{quick_sight_id}", + connection_qualified_name=connection_qualified_name, + connector_name=AtlanConnectorType.get_connector_name( + connection_qualified_name + ), + quick_sight_dashboard_folders=folders, + ) + attributes: QuickSightDashboard.Attributes = Field( default_factory=lambda: QuickSightDashboard.Attributes(), description=( diff --git a/pyatlan/model/assets/quick_sight_dashboard_visual.py b/pyatlan/model/assets/quick_sight_dashboard_visual.py index 43db41bd7..29324002f 100644 --- a/pyatlan/model/assets/quick_sight_dashboard_visual.py +++ b/pyatlan/model/assets/quick_sight_dashboard_visual.py @@ -4,11 +4,13 @@ from __future__ import annotations -from typing import ClassVar, List, Optional +from typing import ClassVar, List, Optional, overload from pydantic.v1 import Field, validator +from pyatlan.model.enums import AtlanConnectorType from pyatlan.model.fields.atlan_fields import KeywordTextField, RelationField +from pyatlan.utils import init_guid, validate_required_fields from .quick_sight import QuickSight @@ -16,6 +18,69 @@ class QuickSightDashboardVisual(QuickSight): """Description""" + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_dashboard_qualified_name: str, + ) -> QuickSightDashboardVisual: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_dashboard_qualified_name: str, + connection_qualified_name: str, + ) -> QuickSightDashboardVisual: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_dashboard_qualified_name: str, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightDashboardVisual: + validate_required_fields( + [ + "name", + "quick_sight_id", + "quick_sight_sheet_id", + "quick_sight_sheet_name", + "quick_sight_dashboard_qualified_name", + ], + [ + name, + quick_sight_id, + quick_sight_sheet_id, + quick_sight_sheet_name, + quick_sight_dashboard_qualified_name, + ], + ) + attributes = QuickSightDashboardVisual.Attributes.creator( + name=name, + quick_sight_id=quick_sight_id, + quick_sight_sheet_id=quick_sight_sheet_id, + quick_sight_sheet_name=quick_sight_sheet_name, + quick_sight_dashboard_qualified_name=quick_sight_dashboard_qualified_name, + connection_qualified_name=connection_qualified_name, + ) + return cls(attributes=attributes) + type_name: str = Field(default="QuickSightDashboardVisual", allow_mutation=False) @validator("type_name") @@ -90,6 +155,60 @@ class Attributes(QuickSight.Attributes): default=None, description="" ) # relationship + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_dashboard_qualified_name: str, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightDashboardVisual.Attributes: + validate_required_fields( + [ + "name", + "quick_sight_id", + "quick_sight_sheet_name", + "quick_sight_sheet_id", + "quick_sight_dashboard_qualified_name", + ], + [ + name, + quick_sight_id, + quick_sight_sheet_name, + quick_sight_sheet_id, + quick_sight_dashboard_qualified_name, + ], + ) + assert quick_sight_dashboard_qualified_name + if connection_qualified_name: + connector_name = AtlanConnectorType.get_connector_name( + connection_qualified_name + ) + else: + connection_qn, connector_name = AtlanConnectorType.get_connector_name( + quick_sight_dashboard_qualified_name, + "quick_sight_dashboard_qualified_name", + 4, + ) + connection_qualified_name = connection_qualified_name or connection_qn + return QuickSightDashboardVisual.Attributes( + name=name, + qualified_name=f"{quick_sight_dashboard_qualified_name}/{quick_sight_sheet_id}/{quick_sight_id}", + quick_sight_id=quick_sight_id, + quick_sight_sheet_id=quick_sight_sheet_id, + quick_sight_sheet_name=quick_sight_sheet_name, + quick_sight_dashboard_qualified_name=quick_sight_dashboard_qualified_name, + quick_sight_dashboard=QuickSightDashboard.ref_by_qualified_name( + quick_sight_dashboard_qualified_name + ), + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) + attributes: QuickSightDashboardVisual.Attributes = Field( default_factory=lambda: QuickSightDashboardVisual.Attributes(), description=( diff --git a/pyatlan/model/assets/quick_sight_dataset.py b/pyatlan/model/assets/quick_sight_dataset.py index 1ad49c379..c1a2557ef 100644 --- a/pyatlan/model/assets/quick_sight_dataset.py +++ b/pyatlan/model/assets/quick_sight_dataset.py @@ -4,12 +4,13 @@ from __future__ import annotations -from typing import ClassVar, List, Optional +from typing import ClassVar, List, Optional, overload from pydantic.v1 import Field, validator -from pyatlan.model.enums import QuickSightDatasetImportMode +from pyatlan.model.enums import AtlanConnectorType, QuickSightDatasetImportMode from pyatlan.model.fields.atlan_fields import KeywordField, NumericField, RelationField +from pyatlan.utils import init_guid, validate_required_fields from .quick_sight import QuickSight @@ -17,6 +18,52 @@ class QuickSightDataset(QuickSight): """Description""" + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + ) -> QuickSightDataset: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dataset_import_mode: QuickSightDatasetImportMode, + quick_sight_dataset_folders: List[str], + ) -> QuickSightDataset: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dataset_import_mode: Optional[QuickSightDatasetImportMode] = None, + quick_sight_dataset_folders: Optional[List[str]] = None, + ) -> QuickSightDataset: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + attributes = QuickSightDataset.Attributes.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + quick_sight_dataset_import_mode=quick_sight_dataset_import_mode, + quick_sight_dataset_folders=quick_sight_dataset_folders, + ) + return cls(attributes=attributes) + type_name: str = Field(default="QuickSightDataset", allow_mutation=False) @validator("type_name") @@ -145,6 +192,42 @@ class Attributes(QuickSight.Attributes): default=None, description="" ) # relationship + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_dataset_import_mode: Optional[ + QuickSightDatasetImportMode + ] = None, + quick_sight_dataset_folders: Optional[List[str]] = None, + ) -> QuickSightDataset.Attributes: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + folders = None + if quick_sight_dataset_folders: + folders = [ + QuickSightFolder.ref_by_qualified_name(quick_sight_folder_qn) + for quick_sight_folder_qn in quick_sight_dataset_folders + ] + + return QuickSightDataset.Attributes( + name=name, + quick_sight_id=quick_sight_id, + qualified_name=f"{connection_qualified_name}/{quick_sight_id}", + connection_qualified_name=connection_qualified_name, + connector_name=AtlanConnectorType.get_connector_name( + connection_qualified_name + ), + quick_sight_dataset_import_mode=quick_sight_dataset_import_mode, + quick_sight_dataset_folders=folders, + ) + attributes: QuickSightDataset.Attributes = Field( default_factory=lambda: QuickSightDataset.Attributes(), description=( diff --git a/pyatlan/model/assets/quick_sight_dataset_field.py b/pyatlan/model/assets/quick_sight_dataset_field.py index e9402b5f0..b3a1c7f5b 100644 --- a/pyatlan/model/assets/quick_sight_dataset_field.py +++ b/pyatlan/model/assets/quick_sight_dataset_field.py @@ -4,16 +4,17 @@ from __future__ import annotations -from typing import ClassVar, List, Optional +from typing import ClassVar, List, Optional, overload from pydantic.v1 import Field, validator -from pyatlan.model.enums import QuickSightDatasetFieldType +from pyatlan.model.enums import AtlanConnectorType, QuickSightDatasetFieldType from pyatlan.model.fields.atlan_fields import ( KeywordField, KeywordTextField, RelationField, ) +from pyatlan.utils import init_guid, validate_required_fields from .quick_sight import QuickSight @@ -21,6 +22,52 @@ class QuickSightDatasetField(QuickSight): """Description""" + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_dataset_qualified_name: str, + quick_sight_id: str, + ) -> QuickSightDatasetField: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + quick_sight_dataset_qualified_name: str, + quick_sight_id: str, + quick_sight_dataset_field_type: QuickSightDatasetFieldType, + connection_qualified_name: str, + ) -> QuickSightDatasetField: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_id: str, + quick_sight_dataset_qualified_name: str, + quick_sight_dataset_field_type: Optional[QuickSightDatasetFieldType] = None, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightDatasetField: + validate_required_fields( + ["name", "quick_sight_id", "quick_sight_dataset_qualified_name"], + [name, quick_sight_id, quick_sight_dataset_qualified_name], + ) + attributes = QuickSightDatasetField.Attributes.creator( + name=name, + quick_sight_id=quick_sight_id, + connection_qualified_name=connection_qualified_name, + quick_sight_dataset_qualified_name=quick_sight_dataset_qualified_name, + quick_sight_dataset_field_type=quick_sight_dataset_field_type, + ) + return cls(attributes=attributes) + type_name: str = Field(default="QuickSightDatasetField", allow_mutation=False) @validator("type_name") @@ -115,6 +162,46 @@ class Attributes(QuickSight.Attributes): default=None, description="" ) # relationship + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + quick_sight_dataset_qualified_name: str, + quick_sight_id: str, + quick_sight_dataset_field_type: Optional[QuickSightDatasetFieldType] = None, + connection_qualified_name: Optional[str] = None, + ) -> QuickSightDatasetField.Attributes: + validate_required_fields( + ["name", "quick_sight_id", "quick_sight_dataset_qualified_name"], + [name, quick_sight_id, quick_sight_dataset_qualified_name], + ) + assert quick_sight_dataset_qualified_name + if connection_qualified_name: + connector_name = AtlanConnectorType.get_connector_name( + connection_qualified_name + ) + else: + connection_qn, connector_name = AtlanConnectorType.get_connector_name( + quick_sight_dataset_qualified_name, + "quick_sight_dataset_qualified_name", + 4, + ) + connection_qualified_name = connection_qualified_name or connection_qn + return QuickSightDatasetField.Attributes( + name=name, + quick_sight_dataset_qualified_name=quick_sight_dataset_qualified_name, + quick_sight_id=quick_sight_id, + quick_sight_dataset=QuickSightDataset.ref_by_qualified_name( + quick_sight_dataset_qualified_name + ), + qualified_name=f"{quick_sight_dataset_qualified_name}/{quick_sight_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + quick_sight_dataset_field_type=quick_sight_dataset_field_type, + ) + attributes: QuickSightDatasetField.Attributes = Field( default_factory=lambda: QuickSightDatasetField.Attributes(), description=( diff --git a/pyatlan/model/assets/quick_sight_folder.py b/pyatlan/model/assets/quick_sight_folder.py index 2087f5e97..1fae18349 100644 --- a/pyatlan/model/assets/quick_sight_folder.py +++ b/pyatlan/model/assets/quick_sight_folder.py @@ -4,12 +4,13 @@ from __future__ import annotations -from typing import ClassVar, Dict, List, Optional +from typing import ClassVar, Dict, List, Optional, overload from pydantic.v1 import Field, validator -from pyatlan.model.enums import QuickSightFolderType +from pyatlan.model.enums import AtlanConnectorType, QuickSightFolderType from pyatlan.model.fields.atlan_fields import KeywordField, RelationField +from pyatlan.utils import init_guid, validate_required_fields from .quick_sight import QuickSight @@ -17,6 +18,49 @@ class QuickSightFolder(QuickSight): """Description""" + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + ) -> QuickSightFolder: ... + + @overload + @classmethod + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_folder_type: QuickSightFolderType, + ) -> QuickSightFolder: ... + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_folder_type: Optional[QuickSightFolderType] = None, + ) -> QuickSightFolder: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + attributes = QuickSightFolder.Attributes.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + quick_sight_folder_type=quick_sight_folder_type, + ) + return cls(attributes=attributes) + type_name: str = Field(default="QuickSightFolder", allow_mutation=False) @validator("type_name") @@ -151,6 +195,31 @@ class Attributes(QuickSight.Attributes): default=None, description="" ) # relationship + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + quick_sight_id: str, + quick_sight_folder_type: Optional[QuickSightFolderType] = None, + ) -> QuickSightFolder.Attributes: + validate_required_fields( + ["name", "connection_qualified_name", "quick_sight_id"], + [name, connection_qualified_name, quick_sight_id], + ) + return QuickSightFolder.Attributes( + name=name, + quick_sight_id=quick_sight_id, + qualified_name=f"{connection_qualified_name}/{quick_sight_id}", + connection_qualified_name=connection_qualified_name, + connector_name=AtlanConnectorType.get_connector_name( + connection_qualified_name + ), + quick_sight_folder_type=quick_sight_folder_type, + ) + attributes: QuickSightFolder.Attributes = Field( default_factory=lambda: QuickSightFolder.Attributes(), description=( diff --git a/tests/integration/adls_asset_test.py b/tests/integration/adls_asset_test.py index bb67d8bfe..1644946f1 100644 --- a/tests/integration/adls_asset_test.py +++ b/tests/integration/adls_asset_test.py @@ -11,11 +11,9 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from pyatlan.utils import get_parent_qualified_name from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("ADLS") @@ -33,9 +31,6 @@ ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: result = create_connection( diff --git a/tests/integration/airflow_asset_test.py b/tests/integration/airflow_asset_test.py index a6608102d..7dae854cf 100644 --- a/tests/integration/airflow_asset_test.py +++ b/tests/integration/airflow_asset_test.py @@ -13,10 +13,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("AIRFLOW") @@ -31,9 +29,6 @@ ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: result = create_connection( diff --git a/tests/integration/anaplan_asset_test.py b/tests/integration/anaplan_asset_test.py index c6c762530..ff4689626 100644 --- a/tests/integration/anaplan_asset_test.py +++ b/tests/integration/anaplan_asset_test.py @@ -23,10 +23,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("ANAPLAN") @@ -56,9 +54,6 @@ ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: result = create_connection( diff --git a/tests/integration/api_asset_test.py b/tests/integration/api_asset_test.py index dd042a6b4..b5471c5f4 100644 --- a/tests/integration/api_asset_test.py +++ b/tests/integration/api_asset_test.py @@ -19,10 +19,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("API") @@ -60,9 +58,6 @@ API_FIELD_REFERENCE_OBJECT_NAME = f"{MODULE_NAME}-api-object-reference" -response = block(AtlanClient(), AssetMutationResponse()) - - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: result = create_connection( diff --git a/tests/integration/app_asset_test.py b/tests/integration/app_asset_test.py index 1bddfef69..1e722c6f4 100644 --- a/tests/integration/app_asset_test.py +++ b/tests/integration/app_asset_test.py @@ -11,10 +11,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("APP") @@ -29,9 +27,6 @@ ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: result = create_connection( diff --git a/tests/integration/azure_event_hub_asset_test.py b/tests/integration/azure_event_hub_asset_test.py index 04dc7b343..152d62181 100644 --- a/tests/integration/azure_event_hub_asset_test.py +++ b/tests/integration/azure_event_hub_asset_test.py @@ -14,10 +14,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("AZURE_EVENT_HUB") @@ -31,9 +29,6 @@ ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: result = create_connection( diff --git a/tests/integration/custom_asset_test.py b/tests/integration/custom_asset_test.py index 380f90cd8..2d47e3bab 100644 --- a/tests/integration/custom_asset_test.py +++ b/tests/integration/custom_asset_test.py @@ -11,10 +11,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("CUSTOM") @@ -27,8 +25,6 @@ CERTIFICATE_MESSAGE = "Automated testing of the Python SDK." ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: diff --git a/tests/integration/data_mesh_test.py b/tests/integration/data_mesh_test.py index 2c2dabb42..33de2e708 100644 --- a/tests/integration/data_mesh_test.py +++ b/tests/integration/data_mesh_test.py @@ -25,11 +25,10 @@ EntityStatus, ) from pyatlan.model.fluent_search import FluentSearch -from pyatlan.model.response import AssetMutationResponse from pyatlan.model.typedef import AttributeDef, CustomMetadataDef from tests.integration.client import TestId, delete_asset from tests.integration.custom_metadata_test import create_custom_metadata -from tests.integration.utils import block, wait_for_successful_custometadatadef_purge +from tests.integration.utils import wait_for_successful_custometadatadef_purge DATA_PRODUCT_ASSETS_PLAYBOOK_FILTER = ( '{"condition":"AND","isGroupLocked":false,"rules":[]}' @@ -59,8 +58,6 @@ ANNOUNCEMENT_TITLE = "Python SDK testing." ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - @pytest.fixture(scope="module") def domain(client: AtlanClient) -> Generator[DataDomain, None, None]: diff --git a/tests/integration/data_studio_asset_test.py b/tests/integration/data_studio_asset_test.py index 56b3a51eb..9e01a267d 100644 --- a/tests/integration/data_studio_asset_test.py +++ b/tests/integration/data_studio_asset_test.py @@ -12,10 +12,8 @@ EntityStatus, GoogleDatastudioAssetType, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("datastudio") @@ -28,8 +26,6 @@ ANNOUNCEMENT_TITLE = "Python SDK testing." ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: diff --git a/tests/integration/gcs_asset_test.py b/tests/integration/gcs_asset_test.py index 122664f0b..d248820f9 100644 --- a/tests/integration/gcs_asset_test.py +++ b/tests/integration/gcs_asset_test.py @@ -11,10 +11,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("GCS") @@ -28,8 +26,6 @@ ANNOUNCEMENT_TITLE = "Python SDK testing." ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: diff --git a/tests/integration/insights_test.py b/tests/integration/insights_test.py index 7847e66eb..1fffc8b3e 100644 --- a/tests/integration/insights_test.py +++ b/tests/integration/insights_test.py @@ -8,9 +8,7 @@ from pyatlan.model.assets import Collection, Folder, Query, Schema from pyatlan.model.enums import AtlanConnectorType, EntityStatus from pyatlan.model.fluent_search import FluentSearch -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset -from tests.integration.utils import block PREFIX = TestId.make_unique("INS") @@ -25,8 +23,6 @@ SCHEMA_NAME = "WIDE_WORLD_IMPORTERS" USER_DESCRIPTION = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - @pytest.fixture(scope="module") def collection(client: AtlanClient) -> Generator[Collection, None, None]: diff --git a/tests/integration/kafka_asset_test.py b/tests/integration/kafka_asset_test.py index f49f7f6bb..ca3a0b87d 100644 --- a/tests/integration/kafka_asset_test.py +++ b/tests/integration/kafka_asset_test.py @@ -14,10 +14,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("KAFKA") @@ -31,9 +29,6 @@ ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: result = create_connection( diff --git a/tests/integration/preset_asset_test.py b/tests/integration/preset_asset_test.py index 242846916..cf66f00df 100644 --- a/tests/integration/preset_asset_test.py +++ b/tests/integration/preset_asset_test.py @@ -17,10 +17,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("PRESET") @@ -38,8 +36,6 @@ ANNOUNCEMENT_TITLE = "Python SDK testing." ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: diff --git a/tests/integration/quick_sight_asset_test.py b/tests/integration/quick_sight_asset_test.py new file mode 100644 index 000000000..6a04dfe55 --- /dev/null +++ b/tests/integration/quick_sight_asset_test.py @@ -0,0 +1,440 @@ +from typing import Generator + +import pytest + +from pyatlan.client.atlan import AtlanClient +from pyatlan.model.assets import ( + Connection, + QuickSightAnalysis, + QuickSightAnalysisVisual, + QuickSightDashboard, + QuickSightDashboardVisual, + QuickSightDataset, + QuickSightDatasetField, + QuickSightFolder, +) +from pyatlan.model.enums import ( + AtlanConnectorType, + QuickSightDatasetFieldType, + QuickSightDatasetImportMode, + QuickSightFolderType, +) +from tests.integration.client import TestId, delete_asset +from tests.integration.connection_test import create_connection + +MODULE_NAME = TestId.make_unique("QUICKSIGHT") + +CONNECTOR_TYPE = AtlanConnectorType.QUICKSIGHT +QUICKSIGHT_FOLDER_NAME = f"{MODULE_NAME}-QUICKSIGHT-FOLDER" +QUICKSIGHT_FOLDER_ID = f"{MODULE_NAME}-FOLDER-ID" +QUICKSIGHT_DATASET_NAME = f"{MODULE_NAME}-QUICKSIGHT-DATASET" +QUICKSIGHT_DATASET_ID = f"{MODULE_NAME}-DATASET-ID" +QUICKSIGHT_DASHBOARD_NAME = f"{MODULE_NAME}-QUICKSIGHT-DASHBOARD" +QUICKSIGHT_DASHBOARD_ID = f"{MODULE_NAME}-DASHBOARD-ID" +QUICKSIGHT_ANALYSIS_NAME = f"{MODULE_NAME}-QUICKSIGHT-ANALYSIS" +QUICKSIGHT_ANALYSIS_ID = f"{MODULE_NAME}-ANALYSIS-ID" +QUICKSIGHT_DATASET_FIELD_NAME = f"{MODULE_NAME}-QUICKSIGHT-DATASET-FIELD" +QUICKSIGHT_DATASET_FIELD_ID = f"{MODULE_NAME}-DATASET-FIELD-ID" +QUICKSIGHT_DASHBOARD_VISUAL_NAME = f"{MODULE_NAME}-QUICKSIGHT-DASHBOARD-VISUAL" +QUICKSIGHT_DASHBOARD_VISUAL_ID = f"{MODULE_NAME}-DASHBOARD-VISUAL-ID" +QUICKSIGHT_ANALYSIS_VISUAL_NAME = f"{MODULE_NAME}-QUICKSIGHT-ANALYSIS-VISUAL" +QUICKSIGHT_ANALYSIS_VISUAL_ID = f"{MODULE_NAME}-ANALYSIS-VISUAL-ID" +QUICKSIGHT_SHEET_NAME = f"{MODULE_NAME}-QUICKSIGHT-SHEET-NAME" +QUICKSIGHT_SHEET_ID = f"{MODULE_NAME}-SHEET-ID" +QUICK_SIGHT_DESCRIPTION = "Automated testing of the Python SDK." + + +@pytest.fixture(scope="module") +def connection(client: AtlanClient) -> Generator[Connection, None, None]: + result = create_connection( + client=client, name=MODULE_NAME, connector_type=CONNECTOR_TYPE + ) + yield result + delete_asset(client, guid=result.guid, asset_type=Connection) + + +@pytest.fixture(scope="module") +def quick_sight_folder( + client: AtlanClient, connection: Connection +) -> Generator[QuickSightFolder, None, None]: + assert connection.qualified_name + to_create = QuickSightFolder.creator( + name=QUICKSIGHT_FOLDER_NAME, + connection_qualified_name=connection.qualified_name, + quick_sight_id=QUICKSIGHT_FOLDER_ID, + quick_sight_folder_type=QuickSightFolderType.SHARED, + ) + response = client.asset.save(to_create) + result = response.assets_created(asset_type=QuickSightFolder)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=QuickSightFolder) + + +@pytest.fixture(scope="module") +def quick_sight_dataset( + client: AtlanClient, connection: Connection, quick_sight_folder: QuickSightFolder +) -> Generator[QuickSightDataset, None, None]: + assert connection.qualified_name + assert quick_sight_folder.qualified_name + to_create = QuickSightDataset.creator( + name=QUICKSIGHT_DATASET_NAME, + connection_qualified_name=connection.qualified_name, + quick_sight_id=QUICKSIGHT_DATASET_ID, + quick_sight_dataset_import_mode=QuickSightDatasetImportMode.SPICE, + quick_sight_dataset_folders=[str(quick_sight_folder.qualified_name)], + ) + response = client.asset.save(to_create) + result = response.assets_created(asset_type=QuickSightDataset)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=QuickSightDataset) + + +@pytest.fixture(scope="module") +def quick_sight_dashboard( + client: AtlanClient, connection: Connection, quick_sight_folder: QuickSightFolder +) -> Generator[QuickSightDashboard, None, None]: + assert connection.qualified_name + assert quick_sight_folder.qualified_name + to_create = QuickSightDashboard.creator( + name=QUICKSIGHT_DASHBOARD_NAME, + connection_qualified_name=connection.qualified_name, + quick_sight_id=QUICKSIGHT_DASHBOARD_ID, + quick_sight_dashboard_folders=[str(quick_sight_folder.qualified_name)], + ) + response = client.asset.save(to_create) + result = response.assets_created(asset_type=QuickSightDashboard)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=QuickSightDashboard) + + +@pytest.fixture(scope="module") +def quick_sight_analysis( + client: AtlanClient, connection: Connection, quick_sight_folder: QuickSightFolder +) -> Generator[QuickSightAnalysis, None, None]: + assert connection.qualified_name + assert quick_sight_folder.qualified_name + to_create = QuickSightAnalysis.creator( + name=QUICKSIGHT_ANALYSIS_NAME, + connection_qualified_name=connection.qualified_name, + quick_sight_id=QUICKSIGHT_ANALYSIS_ID, + quick_sight_analysis_folders=[str(quick_sight_folder.qualified_name)], + ) + response = client.asset.save(to_create) + result = response.assets_created(asset_type=QuickSightAnalysis)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=QuickSightAnalysis) + + +@pytest.fixture(scope="module") +def quick_sight_dataset_field( + client: AtlanClient, + connection: Connection, + quick_sight_dataset: QuickSightDataset, +) -> Generator[QuickSightDatasetField, None, None]: + assert connection.qualified_name + to_create = QuickSightDatasetField.creator( + name=QUICKSIGHT_DATASET_FIELD_NAME, + quick_sight_dataset_qualified_name=str(quick_sight_dataset.qualified_name), + connection_qualified_name=connection.qualified_name, + quick_sight_id=QUICKSIGHT_DATASET_FIELD_ID, + quick_sight_dataset_field_type=QuickSightDatasetFieldType.STRING, + ) + response = client.asset.save(to_create) + result = response.assets_created(asset_type=QuickSightDatasetField)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=QuickSightDatasetField) + + +@pytest.fixture(scope="module") +def quick_sight_analysis_visual( + client: AtlanClient, + connection: Connection, + quick_sight_analysis: QuickSightAnalysis, +) -> Generator[QuickSightAnalysisVisual, None, None]: + assert connection.qualified_name + to_create = QuickSightAnalysisVisual.creator( + name=QUICKSIGHT_ANALYSIS_VISUAL_NAME, + quick_sight_sheet_id=QUICKSIGHT_SHEET_ID, + quick_sight_sheet_name=QUICKSIGHT_SHEET_NAME, + quick_sight_analysis_qualified_name=str(quick_sight_analysis.qualified_name), + connection_qualified_name=connection.qualified_name, + quick_sight_id=QUICKSIGHT_ANALYSIS_VISUAL_ID, + ) + response = client.asset.save(to_create) + result = response.assets_created(asset_type=QuickSightAnalysisVisual)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=QuickSightAnalysisVisual) + + +@pytest.fixture(scope="module") +def quick_sight_dashboard_visual( + client: AtlanClient, + connection: Connection, + quick_sight_dashboard: QuickSightDashboard, +) -> Generator[QuickSightDashboardVisual, None, None]: + assert connection.qualified_name + to_create = QuickSightDashboardVisual.creator( + name=QUICKSIGHT_DASHBOARD_VISUAL_NAME, + quick_sight_sheet_id=QUICKSIGHT_SHEET_ID, + quick_sight_sheet_name=QUICKSIGHT_SHEET_NAME, + quick_sight_dashboard_qualified_name=str(quick_sight_dashboard.qualified_name), + connection_qualified_name=connection.qualified_name, + quick_sight_id=QUICKSIGHT_DASHBOARD_VISUAL_ID, + ) + response = client.asset.save(to_create) + result = response.assets_created(asset_type=QuickSightDashboardVisual)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=QuickSightDashboardVisual) + + +def test_sight_folder( + client: AtlanClient, connection: Connection, quick_sight_folder: QuickSightFolder +): + assert quick_sight_folder + assert quick_sight_folder.guid + assert quick_sight_folder.qualified_name + assert quick_sight_folder.name == QUICKSIGHT_FOLDER_NAME + assert quick_sight_folder.quick_sight_id == QUICKSIGHT_FOLDER_ID + assert quick_sight_folder.connection_qualified_name == connection.qualified_name + assert quick_sight_folder.connector_name == AtlanConnectorType.QUICKSIGHT.value + assert quick_sight_folder.quick_sight_folder_type == QuickSightFolderType.SHARED + + to_update = quick_sight_folder.updater( + name=quick_sight_folder.name, qualified_name=quick_sight_folder.qualified_name + ) + to_update.description = QUICK_SIGHT_DESCRIPTION + response = client.asset.save(to_update) + assert response and response.mutated_entities + + asset = client.asset.get_by_qualified_name( + qualified_name=quick_sight_folder.qualified_name, asset_type=QuickSightFolder + ) + assert asset + assert asset.name == QUICKSIGHT_FOLDER_NAME + assert asset.description == QUICK_SIGHT_DESCRIPTION + assert asset.qualified_name == quick_sight_folder.qualified_name + + +def test_sight_dataset( + client: AtlanClient, connection: Connection, quick_sight_dataset: QuickSightDataset +): + assert quick_sight_dataset + assert quick_sight_dataset.guid + assert quick_sight_dataset.qualified_name + assert quick_sight_dataset.name == QUICKSIGHT_DATASET_NAME + assert quick_sight_dataset.quick_sight_id == QUICKSIGHT_DATASET_ID + assert quick_sight_dataset.connection_qualified_name == connection.qualified_name + assert quick_sight_dataset.connector_name == AtlanConnectorType.QUICKSIGHT.value + assert ( + quick_sight_dataset.quick_sight_dataset_import_mode + == QuickSightDatasetImportMode.SPICE + ) + + to_update = quick_sight_dataset.updater( + name=quick_sight_dataset.name, qualified_name=quick_sight_dataset.qualified_name + ) + to_update.description = QUICK_SIGHT_DESCRIPTION + response = client.asset.save(to_update) + assert response and response.mutated_entities + + asset = client.asset.get_by_qualified_name( + qualified_name=quick_sight_dataset.qualified_name, asset_type=QuickSightDataset + ) + assert asset + assert asset.name == QUICKSIGHT_DATASET_NAME + assert asset.description == QUICK_SIGHT_DESCRIPTION + assert asset.qualified_name == quick_sight_dataset.qualified_name + + +def test_sight_dashboard( + client: AtlanClient, + connection: Connection, + quick_sight_dashboard: QuickSightDashboard, +): + assert quick_sight_dashboard + assert quick_sight_dashboard.guid + assert quick_sight_dashboard.qualified_name + assert quick_sight_dashboard.name == QUICKSIGHT_DASHBOARD_NAME + assert quick_sight_dashboard.quick_sight_id == QUICKSIGHT_DASHBOARD_ID + assert quick_sight_dashboard.connection_qualified_name == connection.qualified_name + assert quick_sight_dashboard.connector_name == AtlanConnectorType.QUICKSIGHT.value + + to_update = quick_sight_dashboard.updater( + name=quick_sight_dashboard.name, + qualified_name=quick_sight_dashboard.qualified_name, + ) + to_update.description = QUICK_SIGHT_DESCRIPTION + response = client.asset.save(to_update) + assert response and response.mutated_entities + + asset = client.asset.get_by_qualified_name( + qualified_name=quick_sight_dashboard.qualified_name, + asset_type=QuickSightDashboard, + ) + assert asset + assert asset.name == QUICKSIGHT_DASHBOARD_NAME + assert asset.description == QUICK_SIGHT_DESCRIPTION + assert asset.qualified_name == quick_sight_dashboard.qualified_name + + +def test_sight_analysis( + client: AtlanClient, + connection: Connection, + quick_sight_analysis: QuickSightAnalysis, +): + assert quick_sight_analysis + assert quick_sight_analysis.guid + assert quick_sight_analysis.qualified_name + assert quick_sight_analysis.name == QUICKSIGHT_ANALYSIS_NAME + assert quick_sight_analysis.quick_sight_id == QUICKSIGHT_ANALYSIS_ID + assert quick_sight_analysis.connection_qualified_name == connection.qualified_name + assert quick_sight_analysis.connector_name == AtlanConnectorType.QUICKSIGHT.value + + to_update = quick_sight_analysis.updater( + name=quick_sight_analysis.name, + qualified_name=quick_sight_analysis.qualified_name, + ) + to_update.description = QUICK_SIGHT_DESCRIPTION + response = client.asset.save(to_update) + assert response and response.mutated_entities + + asset = client.asset.get_by_qualified_name( + qualified_name=quick_sight_analysis.qualified_name, + asset_type=QuickSightAnalysis, + ) + assert asset + assert asset.name == QUICKSIGHT_ANALYSIS_NAME + assert asset.description == QUICK_SIGHT_DESCRIPTION + assert asset.qualified_name == quick_sight_analysis.qualified_name + + +def test_sight_dataset_field( + client: AtlanClient, + connection: Connection, + quick_sight_dataset_field: QuickSightDatasetField, + quick_sight_dataset: QuickSightDataset, +): + assert quick_sight_dataset_field + assert quick_sight_dataset_field.guid + assert quick_sight_dataset_field.qualified_name + assert quick_sight_dataset_field.name == QUICKSIGHT_DATASET_FIELD_NAME + assert quick_sight_dataset_field.quick_sight_id == QUICKSIGHT_DATASET_FIELD_ID + assert ( + quick_sight_dataset_field.connection_qualified_name == connection.qualified_name + ) + assert ( + quick_sight_dataset_field.connector_name == AtlanConnectorType.QUICKSIGHT.value + ) + assert ( + quick_sight_dataset_field.quick_sight_dataset_qualified_name + == quick_sight_dataset.qualified_name + ) + assert ( + quick_sight_dataset_field.quick_sight_dataset_field_type + == QuickSightDatasetFieldType.STRING + ) + + to_update = quick_sight_dataset_field.updater( + name=quick_sight_dataset_field.name, + qualified_name=quick_sight_dataset_field.qualified_name, + ) + to_update.description = QUICK_SIGHT_DESCRIPTION + response = client.asset.save(to_update) + assert response and response.mutated_entities + + asset = client.asset.get_by_qualified_name( + qualified_name=quick_sight_dataset_field.qualified_name, + asset_type=QuickSightDatasetField, + ) + assert asset + assert asset.name == QUICKSIGHT_DATASET_FIELD_NAME + assert asset.description == QUICK_SIGHT_DESCRIPTION + assert asset.qualified_name == quick_sight_dataset_field.qualified_name + + +def test_sight_analysis_visual( + client: AtlanClient, + connection: Connection, + quick_sight_analysis_visual: QuickSightAnalysisVisual, + quick_sight_analysis: QuickSightAnalysis, +): + assert quick_sight_analysis_visual + assert quick_sight_analysis_visual.guid + assert quick_sight_analysis_visual.qualified_name + assert quick_sight_analysis_visual.name == QUICKSIGHT_ANALYSIS_VISUAL_NAME + assert quick_sight_analysis_visual.quick_sight_id == QUICKSIGHT_ANALYSIS_VISUAL_ID + assert ( + quick_sight_analysis_visual.connection_qualified_name + == connection.qualified_name + ) + assert ( + quick_sight_analysis_visual.connector_name + == AtlanConnectorType.QUICKSIGHT.value + ) + assert ( + quick_sight_analysis_visual.quick_sight_analysis_qualified_name + == quick_sight_analysis.qualified_name + ) + assert quick_sight_analysis_visual.quick_sight_sheet_id == QUICKSIGHT_SHEET_ID + assert quick_sight_analysis_visual.quick_sight_sheet_name == QUICKSIGHT_SHEET_NAME + + to_update = quick_sight_analysis_visual.updater( + name=quick_sight_analysis_visual.name, + qualified_name=quick_sight_analysis_visual.qualified_name, + ) + to_update.description = QUICK_SIGHT_DESCRIPTION + response = client.asset.save(to_update) + assert response and response.mutated_entities + + asset = client.asset.get_by_qualified_name( + qualified_name=quick_sight_analysis_visual.qualified_name, + asset_type=QuickSightAnalysisVisual, + ) + assert asset + assert asset.name == QUICKSIGHT_ANALYSIS_VISUAL_NAME + assert asset.description == QUICK_SIGHT_DESCRIPTION + assert asset.qualified_name == quick_sight_analysis_visual.qualified_name + + +def test_sight_dashboard_visual( + client: AtlanClient, + connection: Connection, + quick_sight_dashboard_visual: QuickSightDashboardVisual, + quick_sight_dashboard: QuickSightDashboard, +): + assert quick_sight_dashboard_visual + assert quick_sight_dashboard_visual.guid + assert quick_sight_dashboard_visual.qualified_name + assert quick_sight_dashboard_visual.name == QUICKSIGHT_DASHBOARD_VISUAL_NAME + assert quick_sight_dashboard_visual.quick_sight_id == QUICKSIGHT_DASHBOARD_VISUAL_ID + assert ( + quick_sight_dashboard_visual.connection_qualified_name + == connection.qualified_name + ) + assert ( + quick_sight_dashboard_visual.connector_name + == AtlanConnectorType.QUICKSIGHT.value + ) + assert ( + quick_sight_dashboard_visual.quick_sight_dashboard_qualified_name + == quick_sight_dashboard.qualified_name + ) + assert quick_sight_dashboard_visual.quick_sight_sheet_id == QUICKSIGHT_SHEET_ID + assert quick_sight_dashboard_visual.quick_sight_sheet_name == QUICKSIGHT_SHEET_NAME + + to_update = quick_sight_dashboard_visual.updater( + name=quick_sight_dashboard_visual.name, + qualified_name=quick_sight_dashboard_visual.qualified_name, + ) + to_update.description = QUICK_SIGHT_DESCRIPTION + response = client.asset.save(to_update) + assert response and response.mutated_entities + + asset = client.asset.get_by_qualified_name( + qualified_name=quick_sight_dashboard_visual.qualified_name, + asset_type=QuickSightDashboardVisual, + ) + assert asset + assert asset.name == QUICKSIGHT_DASHBOARD_VISUAL_NAME + assert asset.description == QUICK_SIGHT_DESCRIPTION + assert asset.qualified_name == quick_sight_dashboard_visual.qualified_name diff --git a/tests/integration/s3_asset_test.py b/tests/integration/s3_asset_test.py index d6c81a535..662433723 100644 --- a/tests/integration/s3_asset_test.py +++ b/tests/integration/s3_asset_test.py @@ -13,10 +13,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("S3") @@ -34,9 +32,6 @@ ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: result = create_connection( diff --git a/tests/integration/superset_asset_test.py b/tests/integration/superset_asset_test.py index 6113186b7..987d32fde 100644 --- a/tests/integration/superset_asset_test.py +++ b/tests/integration/superset_asset_test.py @@ -16,10 +16,8 @@ CertificateStatus, EntityStatus, ) -from pyatlan.model.response import AssetMutationResponse from tests.integration.client import TestId, delete_asset from tests.integration.connection_test import create_connection -from tests.integration.utils import block MODULE_NAME = TestId.make_unique("SUPERSET") @@ -35,8 +33,6 @@ ANNOUNCEMENT_TITLE = "Python SDK testing." ANNOUNCEMENT_MESSAGE = "Automated testing of the Python SDK." -response = block(AtlanClient(), AssetMutationResponse()) - @pytest.fixture(scope="module") def connection(client: AtlanClient) -> Generator[Connection, None, None]: diff --git a/tests/integration/utils.py b/tests/integration/utils.py index b29a30f76..5069e1a44 100644 --- a/tests/integration/utils.py +++ b/tests/integration/utils.py @@ -1,6 +1,3 @@ -import time -from typing import List - from tenacity import ( retry, retry_if_exception_type, @@ -9,48 +6,10 @@ ) from pyatlan.client.atlan import AtlanClient -from pyatlan.errors import AtlanError, ErrorCode, NotFoundError -from pyatlan.model.assets import Asset -from pyatlan.model.enums import EntityStatus -from pyatlan.model.response import AssetMutationResponse +from pyatlan.errors import AtlanError from pyatlan.model.typedef import AtlanTagDef, CustomMetadataDef, EnumDef -def block( - client: AtlanClient, response: AssetMutationResponse -) -> AssetMutationResponse: - if response.mutated_entities and response.mutated_entities.DELETE: - retrieve_and_check_assets(client, response.mutated_entities.DELETE, 0) - return response - - -def retrieve_and_check_assets( - client: AtlanClient, to_check: List[Asset], retry_count: int -): - """ - Retrieve and check the status of a list of assets and retry if needed. - """ - leftovers = [] - for one in to_check: - try: - candidate = client.asset.get_by_guid( - one.guid, asset_type=type(one), ignore_relationships=False - ) - if candidate and candidate.status == EntityStatus.ACTIVE: - leftovers.append(candidate) - except NotFoundError: - # If it is not found, it was successfully deleted (purged), so we - # do not need to look for it any further - print("Asset no longer exists.") - except AtlanError: - leftovers.append(one) - if leftovers: - if retry_count == 20: - raise ErrorCode.RETRY_OVERRUN.exception_with_parameters() - time.sleep(2) - retrieve_and_check_assets(client, leftovers, retry_count + 1) - - @retry( retry=retry_if_exception_type(AtlanError), wait=wait_random_exponential(multiplier=1, max=5), diff --git a/tests/unit/model/constants.py b/tests/unit/model/constants.py index 7cf1bee0d..5651ba1f3 100644 --- a/tests/unit/model/constants.py +++ b/tests/unit/model/constants.py @@ -261,3 +261,30 @@ f"{DATAVERSE_ENTITY_QUALIFIED_NAME}/{DATAVERSE_ATTRIBUTE_NAME}" ) TABLE_PARTITION_NAME = "test-table-partition" +QUICK_SIGHT_NAME = "test-quick-sight-name" +QUICK_SIGHT_CONNECTION_QUALIFIED_NAME = "default/quicksight/123456789" +QUICK_SIGHT_ID = "0123456789" +QUICK_SIGHT_QUALIFIED_NAME = f"{QUICK_SIGHT_CONNECTION_QUALIFIED_NAME}/{QUICK_SIGHT_ID}" +QUICK_SIGHT_CONNECTOR_TYPE = "quicksight" +QUICK_SIGHT_FOLDER_SET = [ + f"{QUICK_SIGHT_CONNECTION_QUALIFIED_NAME}/1235663folder", + f"{QUICK_SIGHT_CONNECTION_QUALIFIED_NAME}/1235663folder", +] +QUICK_SIGHT_ID_DATASET_FEILD = "23045523" +QUICK_SIGHT_DATASET_FIELD_QUALIFIED_NAME = ( + f"{QUICK_SIGHT_QUALIFIED_NAME}/{QUICK_SIGHT_ID_DATASET_FEILD}" +) +QUICK_SIGHT_ID_ANALYSIS_VISUAL = "23045532" +QUICK_SIGHT_SHEET_ID = "1234" +QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME = ( + f"{QUICK_SIGHT_QUALIFIED_NAME}/" + f"{QUICK_SIGHT_SHEET_ID}/" + f"{QUICK_SIGHT_ID_ANALYSIS_VISUAL}" +) +QUICK_SIGHT_SHEET_NAME = "test-qs-sheet-name" +QUICK_SIGHT_ID_DASHBOARD_VISUAL = "230455332" +QUICK_SIGHT_DASHBOARD_VISUAL_QUALIFIED_NAME = ( + f"{QUICK_SIGHT_QUALIFIED_NAME}/" + f"{QUICK_SIGHT_SHEET_ID}/" + f"{QUICK_SIGHT_ID_DASHBOARD_VISUAL}" +) diff --git a/tests/unit/model/quick_sight_analysis_test.py b/tests/unit/model/quick_sight_analysis_test.py new file mode 100644 index 000000000..4bdf34961 --- /dev/null +++ b/tests/unit/model/quick_sight_analysis_test.py @@ -0,0 +1,92 @@ +import pytest + +from pyatlan.model.assets import QuickSightAnalysis +from tests.unit.model.constants import ( + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_CONNECTOR_TYPE, + QUICK_SIGHT_FOLDER_SET, + QUICK_SIGHT_ID, + QUICK_SIGHT_NAME, + QUICK_SIGHT_QUALIFIED_NAME, +) + + +@pytest.mark.parametrize( + "name, connection_qualified_name, quick_sight_id, message", + [ + ( + None, + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_ID, + "name is required", + ), + ( + QUICK_SIGHT_NAME, + None, + QUICK_SIGHT_ID, + "connection_qualified_name is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + None, + "quick_sight_id is required", + ), + ], +) +def test_creator_with_missing_parameters_raise_value_error( + name: str, connection_qualified_name: str, quick_sight_id: str, message: str +): + with pytest.raises(ValueError, match=message): + QuickSightAnalysis.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + ) + + +def test_creator(): + sut = QuickSightAnalysis.creator( + name=QUICK_SIGHT_NAME, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + quick_sight_id=QUICK_SIGHT_ID, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID + assert sut.qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + + +def test_overload_creator(): + sut = QuickSightAnalysis.creator( + name=QUICK_SIGHT_NAME, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + quick_sight_id=QUICK_SIGHT_ID, + quick_sight_analysis_folders=QUICK_SIGHT_FOLDER_SET, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID + assert sut.qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + + +def test_updater(): + sut = QuickSightAnalysis.updater( + qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, name=QUICK_SIGHT_NAME + ) + + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.name == QUICK_SIGHT_NAME + + +def test_trim_to_required(): + sut = QuickSightAnalysis.updater( + name=QUICK_SIGHT_NAME, qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + ).trim_to_required() + + assert sut.name == QUICK_SIGHT_NAME + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME diff --git a/tests/unit/model/quick_sight_analysis_visual_test.py b/tests/unit/model/quick_sight_analysis_visual_test.py new file mode 100644 index 000000000..c709bc2e7 --- /dev/null +++ b/tests/unit/model/quick_sight_analysis_visual_test.py @@ -0,0 +1,132 @@ +import pytest + +from pyatlan.model.assets import QuickSightAnalysisVisual +from tests.unit.model.constants import ( + QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME, + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_CONNECTOR_TYPE, + QUICK_SIGHT_ID_ANALYSIS_VISUAL, + QUICK_SIGHT_NAME, + QUICK_SIGHT_QUALIFIED_NAME, + QUICK_SIGHT_SHEET_ID, + QUICK_SIGHT_SHEET_NAME, +) + + +@pytest.mark.parametrize( + "name, quick_sight_id, quick_sight_sheet_id, quick_sight_sheet_name, quick_sight_analysis_qualified_name, message", + [ + ( + None, + QUICK_SIGHT_ID_ANALYSIS_VISUAL, + QUICK_SIGHT_SHEET_ID, + QUICK_SIGHT_SHEET_NAME, + QUICK_SIGHT_QUALIFIED_NAME, + "name is required", + ), + ( + QUICK_SIGHT_NAME, + None, + QUICK_SIGHT_SHEET_ID, + QUICK_SIGHT_SHEET_NAME, + QUICK_SIGHT_QUALIFIED_NAME, + "quick_sight_id is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_ID_ANALYSIS_VISUAL, + None, + QUICK_SIGHT_SHEET_NAME, + QUICK_SIGHT_QUALIFIED_NAME, + "quick_sight_sheet_id is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_ID_ANALYSIS_VISUAL, + QUICK_SIGHT_SHEET_ID, + None, + QUICK_SIGHT_QUALIFIED_NAME, + "quick_sight_sheet_name is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_ID_ANALYSIS_VISUAL, + QUICK_SIGHT_SHEET_ID, + QUICK_SIGHT_SHEET_NAME, + None, + "quick_sight_analysis_qualified_name is required", + ), + ], +) +def test_creator_with_missing_parameters_raise_value_error( + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_analysis_qualified_name: str, + message: str, +): + with pytest.raises(ValueError, match=message): + QuickSightAnalysisVisual.creator( + name=name, + quick_sight_sheet_id=quick_sight_sheet_id, + quick_sight_id=quick_sight_id, + quick_sight_sheet_name=quick_sight_sheet_name, + quick_sight_analysis_qualified_name=quick_sight_analysis_qualified_name, + ) + + +def test_creator(): + sut = QuickSightAnalysisVisual.creator( + name=QUICK_SIGHT_NAME, + quick_sight_id=QUICK_SIGHT_ID_ANALYSIS_VISUAL, + quick_sight_sheet_id=QUICK_SIGHT_SHEET_ID, + quick_sight_sheet_name=QUICK_SIGHT_SHEET_NAME, + quick_sight_analysis_qualified_name=QUICK_SIGHT_QUALIFIED_NAME, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.quick_sight_analysis_qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID_ANALYSIS_VISUAL + assert sut.qualified_name == QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + assert sut.quick_sight_sheet_id == QUICK_SIGHT_SHEET_ID + assert sut.quick_sight_sheet_name == QUICK_SIGHT_SHEET_NAME + + +def test_overload_creator(): + sut = QuickSightAnalysisVisual.creator( + name=QUICK_SIGHT_NAME, + quick_sight_id=QUICK_SIGHT_ID_ANALYSIS_VISUAL, + quick_sight_sheet_id=QUICK_SIGHT_SHEET_ID, + quick_sight_sheet_name=QUICK_SIGHT_SHEET_NAME, + quick_sight_analysis_qualified_name=QUICK_SIGHT_QUALIFIED_NAME, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.quick_sight_analysis_qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID_ANALYSIS_VISUAL + assert sut.qualified_name == QUICK_SIGHT_ANALYSIS_VISUAL_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + assert sut.quick_sight_sheet_id == QUICK_SIGHT_SHEET_ID + assert sut.quick_sight_sheet_name == QUICK_SIGHT_SHEET_NAME + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + + +def test_updater(): + sut = QuickSightAnalysisVisual.updater( + qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, name=QUICK_SIGHT_NAME + ) + + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.name == QUICK_SIGHT_NAME + + +def test_trim_to_required(): + sut = QuickSightAnalysisVisual.updater( + name=QUICK_SIGHT_NAME, qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + ).trim_to_required() + + assert sut.name == QUICK_SIGHT_NAME + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME diff --git a/tests/unit/model/quick_sight_dashboard_test.py b/tests/unit/model/quick_sight_dashboard_test.py new file mode 100644 index 000000000..85948c424 --- /dev/null +++ b/tests/unit/model/quick_sight_dashboard_test.py @@ -0,0 +1,92 @@ +import pytest + +from pyatlan.model.assets import QuickSightDashboard +from tests.unit.model.constants import ( + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_CONNECTOR_TYPE, + QUICK_SIGHT_FOLDER_SET, + QUICK_SIGHT_ID, + QUICK_SIGHT_NAME, + QUICK_SIGHT_QUALIFIED_NAME, +) + + +@pytest.mark.parametrize( + "name, connection_qualified_name, quick_sight_id, message", + [ + ( + None, + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_ID, + "name is required", + ), + ( + QUICK_SIGHT_NAME, + None, + QUICK_SIGHT_ID, + "connection_qualified_name is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + None, + "quick_sight_id is required", + ), + ], +) +def test_creator_with_missing_parameters_raise_value_error( + name: str, connection_qualified_name: str, quick_sight_id: str, message: str +): + with pytest.raises(ValueError, match=message): + QuickSightDashboard.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + ) + + +def test_creator(): + sut = QuickSightDashboard.creator( + name=QUICK_SIGHT_NAME, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + quick_sight_id=QUICK_SIGHT_ID, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID + assert sut.qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + + +def test_overload_creator(): + sut = QuickSightDashboard.creator( + name=QUICK_SIGHT_NAME, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + quick_sight_id=QUICK_SIGHT_ID, + quick_sight_dashboard_folders=QUICK_SIGHT_FOLDER_SET, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID + assert sut.qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + + +def test_updater(): + sut = QuickSightDashboard.updater( + qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, name=QUICK_SIGHT_NAME + ) + + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.name == QUICK_SIGHT_NAME + + +def test_trim_to_required(): + sut = QuickSightDashboard.updater( + name=QUICK_SIGHT_NAME, qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + ).trim_to_required() + + assert sut.name == QUICK_SIGHT_NAME + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME diff --git a/tests/unit/model/quick_sight_dashboard_visual_test.py b/tests/unit/model/quick_sight_dashboard_visual_test.py new file mode 100644 index 000000000..b54a87719 --- /dev/null +++ b/tests/unit/model/quick_sight_dashboard_visual_test.py @@ -0,0 +1,132 @@ +import pytest + +from pyatlan.model.assets import QuickSightDashboardVisual +from tests.unit.model.constants import ( + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_CONNECTOR_TYPE, + QUICK_SIGHT_DASHBOARD_VISUAL_QUALIFIED_NAME, + QUICK_SIGHT_ID_DASHBOARD_VISUAL, + QUICK_SIGHT_NAME, + QUICK_SIGHT_QUALIFIED_NAME, + QUICK_SIGHT_SHEET_ID, + QUICK_SIGHT_SHEET_NAME, +) + + +@pytest.mark.parametrize( + "name, quick_sight_id, quick_sight_sheet_id, quick_sight_sheet_name, quick_sight_dashboard_qualified_name, message", + [ + ( + None, + QUICK_SIGHT_ID_DASHBOARD_VISUAL, + QUICK_SIGHT_SHEET_ID, + QUICK_SIGHT_SHEET_NAME, + QUICK_SIGHT_QUALIFIED_NAME, + "name is required", + ), + ( + QUICK_SIGHT_NAME, + None, + QUICK_SIGHT_SHEET_ID, + QUICK_SIGHT_SHEET_NAME, + QUICK_SIGHT_QUALIFIED_NAME, + "quick_sight_id is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_ID_DASHBOARD_VISUAL, + None, + QUICK_SIGHT_SHEET_NAME, + QUICK_SIGHT_QUALIFIED_NAME, + "quick_sight_sheet_id is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_ID_DASHBOARD_VISUAL, + QUICK_SIGHT_SHEET_ID, + None, + QUICK_SIGHT_QUALIFIED_NAME, + "quick_sight_sheet_name is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_ID_DASHBOARD_VISUAL, + QUICK_SIGHT_SHEET_ID, + QUICK_SIGHT_SHEET_NAME, + None, + "quick_sight_dashboard_qualified_name is required", + ), + ], +) +def test_creator_with_missing_parameters_raise_value_error( + name: str, + quick_sight_id: str, + quick_sight_sheet_id: str, + quick_sight_sheet_name: str, + quick_sight_dashboard_qualified_name: str, + message: str, +): + with pytest.raises(ValueError, match=message): + QuickSightDashboardVisual.creator( + name=name, + quick_sight_sheet_id=quick_sight_sheet_id, + quick_sight_id=quick_sight_id, + quick_sight_sheet_name=quick_sight_sheet_name, + quick_sight_dashboard_qualified_name=quick_sight_dashboard_qualified_name, + ) + + +def test_creator(): + sut = QuickSightDashboardVisual.creator( + name=QUICK_SIGHT_NAME, + quick_sight_id=QUICK_SIGHT_ID_DASHBOARD_VISUAL, + quick_sight_sheet_id=QUICK_SIGHT_SHEET_ID, + quick_sight_sheet_name=QUICK_SIGHT_SHEET_NAME, + quick_sight_dashboard_qualified_name=QUICK_SIGHT_QUALIFIED_NAME, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.quick_sight_dashboard_qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID_DASHBOARD_VISUAL + assert sut.qualified_name == QUICK_SIGHT_DASHBOARD_VISUAL_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + assert sut.quick_sight_sheet_id == QUICK_SIGHT_SHEET_ID + assert sut.quick_sight_sheet_name == QUICK_SIGHT_SHEET_NAME + + +def test_overload_creator(): + sut = QuickSightDashboardVisual.creator( + name=QUICK_SIGHT_NAME, + quick_sight_id=QUICK_SIGHT_ID_DASHBOARD_VISUAL, + quick_sight_sheet_id=QUICK_SIGHT_SHEET_ID, + quick_sight_sheet_name=QUICK_SIGHT_SHEET_NAME, + quick_sight_dashboard_qualified_name=QUICK_SIGHT_QUALIFIED_NAME, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.quick_sight_dashboard_qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID_DASHBOARD_VISUAL + assert sut.qualified_name == QUICK_SIGHT_DASHBOARD_VISUAL_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + assert sut.quick_sight_sheet_id == QUICK_SIGHT_SHEET_ID + assert sut.quick_sight_sheet_name == QUICK_SIGHT_SHEET_NAME + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + + +def test_updater(): + sut = QuickSightDashboardVisual.updater( + qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, name=QUICK_SIGHT_NAME + ) + + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.name == QUICK_SIGHT_NAME + + +def test_trim_to_required(): + sut = QuickSightDashboardVisual.updater( + name=QUICK_SIGHT_NAME, qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + ).trim_to_required() + + assert sut.name == QUICK_SIGHT_NAME + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME diff --git a/tests/unit/model/quick_sight_dataset_field_test.py b/tests/unit/model/quick_sight_dataset_field_test.py new file mode 100644 index 000000000..ec2110bec --- /dev/null +++ b/tests/unit/model/quick_sight_dataset_field_test.py @@ -0,0 +1,99 @@ +import pytest + +from pyatlan.model.assets import QuickSightDatasetField +from pyatlan.model.enums import QuickSightDatasetFieldType +from tests.unit.model.constants import ( + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_CONNECTOR_TYPE, + QUICK_SIGHT_DATASET_FIELD_QUALIFIED_NAME, + QUICK_SIGHT_ID_DATASET_FEILD, + QUICK_SIGHT_NAME, + QUICK_SIGHT_QUALIFIED_NAME, +) + + +@pytest.mark.parametrize( + "name, quick_sight_dataset_qualified_name, quick_sight_id, message", + [ + ( + None, + QUICK_SIGHT_QUALIFIED_NAME, + QUICK_SIGHT_ID_DATASET_FEILD, + "name is required", + ), + ( + QUICK_SIGHT_NAME, + None, + QUICK_SIGHT_ID_DATASET_FEILD, + "quick_sight_dataset_qualified_name is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_QUALIFIED_NAME, + None, + "quick_sight_id is required", + ), + ], +) +def test_creator_with_missing_parameters_raise_value_error( + name: str, + quick_sight_dataset_qualified_name: str, + quick_sight_id: str, + message: str, +): + with pytest.raises(ValueError, match=message): + QuickSightDatasetField.creator( + name=name, + quick_sight_dataset_qualified_name=quick_sight_dataset_qualified_name, + quick_sight_id=quick_sight_id, + ) + + +def test_creator(): + sut = QuickSightDatasetField.creator( + name=QUICK_SIGHT_NAME, + quick_sight_dataset_qualified_name=QUICK_SIGHT_QUALIFIED_NAME, + quick_sight_id=QUICK_SIGHT_ID_DATASET_FEILD, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.quick_sight_dataset_qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID_DATASET_FEILD + assert sut.qualified_name == QUICK_SIGHT_DATASET_FIELD_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + + +def test_overload_creator(): + sut = QuickSightDatasetField.creator( + name=QUICK_SIGHT_NAME, + quick_sight_dataset_qualified_name=QUICK_SIGHT_QUALIFIED_NAME, + quick_sight_id=QUICK_SIGHT_ID_DATASET_FEILD, + quick_sight_dataset_field_type=QuickSightDatasetFieldType.STRING, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.quick_sight_dataset_qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID_DATASET_FEILD + assert sut.qualified_name == QUICK_SIGHT_DATASET_FIELD_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.quick_sight_dataset_field_type == QuickSightDatasetFieldType.STRING + + +def test_updater(): + sut = QuickSightDatasetField.updater( + qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, name=QUICK_SIGHT_NAME + ) + + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.name == QUICK_SIGHT_NAME + + +def test_trim_to_required(): + sut = QuickSightDatasetField.updater( + name=QUICK_SIGHT_NAME, qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + ).trim_to_required() + + assert sut.name == QUICK_SIGHT_NAME + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME diff --git a/tests/unit/model/quick_sight_dataset_test.py b/tests/unit/model/quick_sight_dataset_test.py new file mode 100644 index 000000000..2a3e64ebd --- /dev/null +++ b/tests/unit/model/quick_sight_dataset_test.py @@ -0,0 +1,97 @@ +import pytest + +from pyatlan.model.assets import QuickSightDataset +from pyatlan.model.enums import QuickSightDatasetImportMode +from tests.unit.model.constants import ( + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_CONNECTOR_TYPE, + QUICK_SIGHT_FOLDER_SET, + QUICK_SIGHT_ID, + QUICK_SIGHT_NAME, + QUICK_SIGHT_QUALIFIED_NAME, +) + + +@pytest.mark.parametrize( + "name, connection_qualified_name, quick_sight_id, message", + [ + ( + None, + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_ID, + "name is required", + ), + ( + QUICK_SIGHT_NAME, + None, + QUICK_SIGHT_ID, + "connection_qualified_name is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + None, + "quick_sight_id is required", + ), + ], +) +def test_creator_with_missing_parameters_raise_value_error( + name: str, connection_qualified_name: str, quick_sight_id: str, message: str +): + with pytest.raises(ValueError, match=message): + QuickSightDataset.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + ) + + +def test_creator(): + sut = QuickSightDataset.creator( + name=QUICK_SIGHT_NAME, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + quick_sight_id=QUICK_SIGHT_ID, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID + assert sut.qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + + +def test_overload_creator(): + sut = QuickSightDataset.creator( + name=QUICK_SIGHT_NAME, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + quick_sight_id=QUICK_SIGHT_ID, + quick_sight_dataset_import_mode=QuickSightDatasetImportMode.DIRECT_QUERY, + quick_sight_dataset_folders=QUICK_SIGHT_FOLDER_SET, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID + assert ( + sut.quick_sight_dataset_import_mode == QuickSightDatasetImportMode.DIRECT_QUERY + ) + assert sut.qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + + +def test_updater(): + sut = QuickSightDataset.updater( + qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, name=QUICK_SIGHT_NAME + ) + + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.name == QUICK_SIGHT_NAME + + +def test_trim_to_required(): + sut = QuickSightDataset.updater( + name=QUICK_SIGHT_NAME, qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + ).trim_to_required() + + assert sut.name == QUICK_SIGHT_NAME + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME diff --git a/tests/unit/model/quick_sight_folder_test.py b/tests/unit/model/quick_sight_folder_test.py new file mode 100644 index 000000000..b00b69b1f --- /dev/null +++ b/tests/unit/model/quick_sight_folder_test.py @@ -0,0 +1,93 @@ +import pytest + +from pyatlan.model.assets import QuickSightFolder +from pyatlan.model.enums import QuickSightFolderType +from tests.unit.model.constants import ( + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_CONNECTOR_TYPE, + QUICK_SIGHT_ID, + QUICK_SIGHT_NAME, + QUICK_SIGHT_QUALIFIED_NAME, +) + + +@pytest.mark.parametrize( + "name, connection_qualified_name, quick_sight_id, message", + [ + ( + None, + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + QUICK_SIGHT_ID, + "name is required", + ), + ( + QUICK_SIGHT_NAME, + None, + QUICK_SIGHT_ID, + "connection_qualified_name is required", + ), + ( + QUICK_SIGHT_NAME, + QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + None, + "quick_sight_id is required", + ), + ], +) +def test_creator_with_missing_parameters_raise_value_error( + name: str, connection_qualified_name: str, quick_sight_id: str, message: str +): + with pytest.raises(ValueError, match=message): + QuickSightFolder.creator( + name=name, + connection_qualified_name=connection_qualified_name, + quick_sight_id=quick_sight_id, + ) + + +def test_creator(): + sut = QuickSightFolder.creator( + name=QUICK_SIGHT_NAME, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + quick_sight_id=QUICK_SIGHT_ID, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID + assert sut.qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + + +def test_overload_creator(): + sut = QuickSightFolder.creator( + name=QUICK_SIGHT_NAME, + connection_qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, + quick_sight_id=QUICK_SIGHT_ID, + quick_sight_folder_type=QuickSightFolderType.SHARED, + ) + + assert sut.name == QUICK_SIGHT_NAME + assert sut.connection_qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.quick_sight_id == QUICK_SIGHT_ID + assert sut.quick_sight_folder_type == QuickSightFolderType.SHARED + assert sut.qualified_name == QUICK_SIGHT_QUALIFIED_NAME + assert sut.connector_name == QUICK_SIGHT_CONNECTOR_TYPE + + +def test_updater(): + sut = QuickSightFolder.updater( + qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME, name=QUICK_SIGHT_NAME + ) + + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + assert sut.name == QUICK_SIGHT_NAME + + +def test_trim_to_required(): + sut = QuickSightFolder.updater( + name=QUICK_SIGHT_NAME, qualified_name=QUICK_SIGHT_CONNECTION_QUALIFIED_NAME + ).trim_to_required() + + assert sut.name == QUICK_SIGHT_NAME + assert sut.qualified_name == QUICK_SIGHT_CONNECTION_QUALIFIED_NAME