Merge pull request #416 from atlanhq/CJ-181

CJ-181 : `Application Container` Creator and Tests (new typedef)

docs/asset/application.rst

@@ -0,0 +1,10 @@
+.. _application:
+
+Application
+===========
+
+.. module:: pyatlan.model.assets
+    :no-index:
+
+.. autoclass:: Application
+    :members:

docs/asset/applicationcontainer.rst

@@ -0,0 +1,10 @@
+.. _applicationcontainer:
+
+ApplicationContainer
+====================
+
+.. module:: pyatlan.model.assets
+    :no-index:
+
+.. autoclass:: ApplicationContainer
+    :members:

docs/assets.rst

@@ -31,6 +31,8 @@ You can interact with all of the following different kinds of assets:
     asset/airflowtask
     asset/anomalo
     asset/anomalocheck
+    asset/application
+    asset/applicationcontainer
     asset/asset
     asset/atlasglossary
     asset/atlasglossarycategory

pyatlan/generator/templates/methods/asset/application_container.jinja2

@@ -0,0 +1,17 @@
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        connection_qualified_name: str,
+    ) -> ApplicationContainer:
+        validate_required_fields(
+            ["name", "connection_qualified_name"], [name, connection_qualified_name]
+        )
+        attributes = ApplicationContainer.Attributes.creator(
+            name=name,
+            connection_qualified_name=connection_qualified_name,
+        )
+        return cls(attributes=attributes)

pyatlan/generator/templates/methods/attribute/application_container.jinja2

@@ -0,0 +1,20 @@
+
+        @classmethod
+        @init_guid
+        def creator(
+            cls,
+            *,
+            name: str,
+            connection_qualified_name: str,
+        ) -> ApplicationContainer.Attributes:
+            validate_required_fields(
+                ["name", "connection_qualified_name"], [name, connection_qualified_name]
+            )
+            return ApplicationContainer.Attributes(
+                name=name,
+                qualified_name=f"{connection_qualified_name}/{name}",
+                connection_qualified_name=connection_qualified_name,
+                connector_name=AtlanConnectorType.get_connector_name(
+                    connection_qualified_name
+                ),
+            )

pyatlan/model/assets/__init__.py

@@ -35,6 +35,7 @@
         "Spark",
         "Tag",
         "SchemaRegistry",
+        "Application",
         "Fivetran",
         "Stakeholder",
         "AirflowDag",
@@ -86,6 +87,7 @@
         "ModelDataModel",
         "SparkJob",
         "SchemaRegistrySubject",
+        "ApplicationContainer",
         "FivetranConnector",
         "AnomaloCheck",
         "MCIncident",

pyatlan/model/assets/__init__.pyi

@@ -32,6 +32,7 @@ __all__ = [
     "Spark",
     "Tag",
     "SchemaRegistry",
+    "Application",
     "Fivetran",
     "Stakeholder",
     "AirflowDag",
@@ -83,6 +84,7 @@ __all__ = [
     "ModelDataModel",
     "SparkJob",
     "SchemaRegistrySubject",
+    "ApplicationContainer",
     "FivetranConnector",
     "AnomaloCheck",
     "MCIncident",
@@ -350,6 +352,8 @@ from .core.airflow_dag import AirflowDag
 from .core.airflow_task import AirflowTask
 from .core.anomalo import Anomalo
 from .core.anomalo_check import AnomaloCheck
+from .core.application import Application
+from .core.application_container import ApplicationContainer
 from .core.asset import Asset
 from .core.atlas_glossary import AtlasGlossary
 from .core.atlas_glossary_category import AtlasGlossaryCategory

pyatlan/model/assets/a_d_l_s.py

@@ -38,6 +38,12 @@ def __setattr__(self, name, value):
     """
     Unique name of the account for this ADLS asset.
     """
+    ASSET_APPLICATION_QUALIFIED_NAME: ClassVar[KeywordField] = KeywordField(
+        "assetApplicationQualifiedName", "assetApplicationQualifiedName"
+    )
+    """
+    Qualified name of the Application Container that contains this asset.
+    """
     AZURE_RESOURCE_ID: ClassVar[KeywordTextField] = KeywordTextField(
         "azureResourceId", "azureResourceId", "azureResourceId.text"
     )
@@ -63,6 +69,7 @@ def __setattr__(self, name, value):
 
     _convenience_properties: ClassVar[List[str]] = [
         "adls_account_qualified_name",
+        "asset_application_qualified_name",
         "azure_resource_id",
         "azure_location",
         "adls_account_secondary_location",
@@ -83,6 +90,24 @@ def adls_account_qualified_name(self, adls_account_qualified_name: Optional[str]
             self.attributes = self.Attributes()
         self.attributes.adls_account_qualified_name = adls_account_qualified_name
 
+    @property
+    def asset_application_qualified_name(self) -> Optional[str]:
+        return (
+            None
+            if self.attributes is None
+            else self.attributes.asset_application_qualified_name
+        )
+
+    @asset_application_qualified_name.setter
+    def asset_application_qualified_name(
+        self, asset_application_qualified_name: Optional[str]
+    ):
+        if self.attributes is None:
+            self.attributes = self.Attributes()
+        self.attributes.asset_application_qualified_name = (
+            asset_application_qualified_name
+        )
+
     @property
     def azure_resource_id(self) -> Optional[str]:
         return None if self.attributes is None else self.attributes.azure_resource_id
@@ -133,6 +158,9 @@ def azure_tags(self, azure_tags: Optional[List[AzureTag]]):
 
     class Attributes(ObjectStore.Attributes):
         adls_account_qualified_name: Optional[str] = Field(default=None, description="")
+        asset_application_qualified_name: Optional[str] = Field(
+            default=None, description=""
+        )
         azure_resource_id: Optional[str] = Field(default=None, description="")
         azure_location: Optional[str] = Field(default=None, description="")
         adls_account_secondary_location: Optional[str] = Field(

pyatlan/model/assets/core/__init__.py

@@ -14,6 +14,8 @@
 from .airflow_task import AirflowTask
 from .anomalo import Anomalo
 from .anomalo_check import AnomaloCheck
+from .application import Application
+from .application_container import ApplicationContainer
 from .asset import Asset
 from .atlas_glossary import AtlasGlossary
 from .atlas_glossary_category import AtlasGlossaryCategory
@@ -140,6 +142,7 @@
 Spark.Attributes.update_forward_refs(**localns)
 Tag.Attributes.update_forward_refs(**localns)
 SchemaRegistry.Attributes.update_forward_refs(**localns)
+Application.Attributes.update_forward_refs(**localns)
 Fivetran.Attributes.update_forward_refs(**localns)
 Stakeholder.Attributes.update_forward_refs(**localns)
 AirflowDag.Attributes.update_forward_refs(**localns)
@@ -191,6 +194,7 @@
 ModelDataModel.Attributes.update_forward_refs(**localns)
 SparkJob.Attributes.update_forward_refs(**localns)
 SchemaRegistrySubject.Attributes.update_forward_refs(**localns)
+ApplicationContainer.Attributes.update_forward_refs(**localns)
 FivetranConnector.Attributes.update_forward_refs(**localns)
 AnomaloCheck.Attributes.update_forward_refs(**localns)
 MCIncident.Attributes.update_forward_refs(**localns)

pyatlan/model/assets/core/application.py

@@ -0,0 +1,63 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2022 Atlan Pte. Ltd.
+
+
+from __future__ import annotations
+
+from typing import ClassVar, List, Optional
+
+from pydantic.v1 import Field, validator
+
+from pyatlan.model.fields.atlan_fields import KeywordField
+
+from .catalog import Catalog
+
+
+class Application(Catalog):
+    """Description"""
+
+    type_name: str = Field(default="Application", allow_mutation=False)
+
+    @validator("type_name")
+    def validate_type_name(cls, v):
+        if v != "Application":
+            raise ValueError("must be Application")
+        return v
+
+    def __setattr__(self, name, value):
+        if name in Application._convenience_properties:
+            return object.__setattr__(self, name, value)
+        super().__setattr__(name, value)
+
+    APPLICATION_ID: ClassVar[KeywordField] = KeywordField(
+        "applicationId", "applicationId"
+    )
+    """
+    Unique identifier for the Application asset from the source system.
+    """
+
+    _convenience_properties: ClassVar[List[str]] = [
+        "application_id",
+    ]
+
+    @property
+    def application_id(self) -> Optional[str]:
+        return None if self.attributes is None else self.attributes.application_id
+
+    @application_id.setter
+    def application_id(self, application_id: Optional[str]):
+        if self.attributes is None:
+            self.attributes = self.Attributes()
+        self.attributes.application_id = application_id
+
+    class Attributes(Catalog.Attributes):
+        application_id: Optional[str] = Field(default=None, description="")
+
+    attributes: Application.Attributes = Field(
+        default_factory=lambda: Application.Attributes(),
+        description=(
+            "Map of attributes in the instance and their values. "
+            "The specific keys of this map will vary by type, "
+            "so are described in the sub-types of this schema."
+        ),
+    )

pyatlan/model/assets/core/application_container.py

@@ -0,0 +1,113 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2022 Atlan Pte. Ltd.
+
+
+from __future__ import annotations
+
+from typing import ClassVar, List, Optional
+
+from pydantic.v1 import Field, validator
+
+from pyatlan.model.enums import AtlanConnectorType
+from pyatlan.model.fields.atlan_fields import RelationField
+from pyatlan.utils import init_guid, validate_required_fields
+
+from .application import Application
+
+
+class ApplicationContainer(Application):
+    """Description"""
+
+    @classmethod
+    @init_guid
+    def creator(
+        cls,
+        *,
+        name: str,
+        connection_qualified_name: str,
+    ) -> ApplicationContainer:
+        validate_required_fields(
+            ["name", "connection_qualified_name"], [name, connection_qualified_name]
+        )
+        attributes = ApplicationContainer.Attributes.creator(
+            name=name,
+            connection_qualified_name=connection_qualified_name,
+        )
+        return cls(attributes=attributes)
+
+    type_name: str = Field(default="ApplicationContainer", allow_mutation=False)
+
+    @validator("type_name")
+    def validate_type_name(cls, v):
+        if v != "ApplicationContainer":
+            raise ValueError("must be ApplicationContainer")
+        return v
+
+    def __setattr__(self, name, value):
+        if name in ApplicationContainer._convenience_properties:
+            return object.__setattr__(self, name, value)
+        super().__setattr__(name, value)
+
+    APPLICATION_OWNED_ASSETS: ClassVar[RelationField] = RelationField(
+        "applicationOwnedAssets"
+    )
+    """
+    TBC
+    """
+
+    _convenience_properties: ClassVar[List[str]] = [
+        "application_owned_assets",
+    ]
+
+    @property
+    def application_owned_assets(self) -> Optional[List[Catalog]]:
+        return (
+            None
+            if self.attributes is None
+            else self.attributes.application_owned_assets
+        )
+
+    @application_owned_assets.setter
+    def application_owned_assets(
+        self, application_owned_assets: Optional[List[Catalog]]
+    ):
+        if self.attributes is None:
+            self.attributes = self.Attributes()
+        self.attributes.application_owned_assets = application_owned_assets
+
+    class Attributes(Application.Attributes):
+        application_owned_assets: Optional[List[Catalog]] = Field(
+            default=None, description=""
+        )  # relationship
+
+        @classmethod
+        @init_guid
+        def creator(
+            cls,
+            *,
+            name: str,
+            connection_qualified_name: str,
+        ) -> ApplicationContainer.Attributes:
+            validate_required_fields(
+                ["name", "connection_qualified_name"], [name, connection_qualified_name]
+            )
+            return ApplicationContainer.Attributes(
+                name=name,
+                qualified_name=f"{connection_qualified_name}/{name}",
+                connection_qualified_name=connection_qualified_name,
+                connector_name=AtlanConnectorType.get_connector_name(
+                    connection_qualified_name
+                ),
+            )
+
+    attributes: ApplicationContainer.Attributes = Field(
+        default_factory=lambda: ApplicationContainer.Attributes(),
+        description=(
+            "Map of attributes in the instance and their values. "
+            "The specific keys of this map will vary by type, "
+            "so are described in the sub-types of this schema."
+        ),
+    )
+
+
+from .catalog import Catalog  # noqa