Skip to content
New issue

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

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

Already on GitHub? Sign in to your account

CXD-321 : ApplicationField typedef addition #481

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions docs/asset/applicationfield.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
.. _applicationfield:

ApplicationField
================

.. module:: pyatlan.model.assets
:no-index:

.. autoclass:: ApplicationField
:members:
1 change: 1 addition & 0 deletions docs/assets.rst
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ You can interact with all of the following different kinds of assets:
asset/anomalocheck
asset/app
asset/application
asset/applicationfield
asset/asset
asset/atlasglossary
asset/atlasglossarycategory
Expand Down
38 changes: 38 additions & 0 deletions pyatlan/generator/templates/methods/asset/application_field.jinja2
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@

@overload
@classmethod
def creator(
cls,
*,
name: str,
application_qualified_name: str,
) -> ApplicationField: ...

@overload
@classmethod
def creator(
cls,
*,
name: str,
application_qualified_name: str,
connection_qualified_name: str,
) -> ApplicationField: ...

@classmethod
@init_guid
def creator(
cls,
*,
name: str,
application_qualified_name: str,
connection_qualified_name: Optional[str] = None,
) -> ApplicationField:
validate_required_fields(
["name", "application_qualified_name"], [name, application_qualified_name]
)
attributes = ApplicationField.Attributes.create(
name=name,
application_qualified_name=application_qualified_name,
connection_qualified_name=connection_qualified_name,
)
return cls(attributes=attributes)
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@

@classmethod
@init_guid
def create(
cls,
*,
name: str,
application_qualified_name: str,
connection_qualified_name: Optional[str] = None,
) -> ApplicationField.Attributes:
validate_required_fields(
["name", "application_qualified_name"],
[name, application_qualified_name],
)
if connection_qualified_name:
connector_name = AtlanConnectorType.get_connector_name(
connection_qualified_name
)
else:
connection_qn, connector_name = AtlanConnectorType.get_connector_name(
application_qualified_name, "application_qualified_name", 4
)

return ApplicationField.Attributes(
name=name,
qualified_name=f"{application_qualified_name}/{name}",
connection_qualified_name=connection_qualified_name or connection_qn,
connector_name=connector_name,
application_parent=Application.ref_by_qualified_name(application_qualified_name),
)
2 changes: 2 additions & 0 deletions pyatlan/model/assets/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
"Stakeholder",
"AirflowDag",
"AirflowTask",
"ApplicationField",
"Application",
"AdfDataflow",
"AdfDataset",
Expand Down Expand Up @@ -149,6 +150,7 @@
"preset": ["Preset"],
"mode": ["Mode"],
"sigma": ["Sigma"],
"app": ["App"],
"anaplan": ["Anaplan"],
"tableau": ["Tableau"],
"looker": ["Looker"],
Expand Down
2 changes: 2 additions & 0 deletions pyatlan/model/assets/__init__.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ __all__ = [
"Stakeholder",
"AirflowDag",
"AirflowTask",
"ApplicationField",
"Application",
"AdfDataflow",
"AdfDataset",
Expand Down Expand Up @@ -380,6 +381,7 @@ from .core.anomalo import Anomalo
from .core.anomalo_check import AnomaloCheck
from .core.app import App
from .core.application import Application
from .core.application_field import ApplicationField
from .core.asset import Asset
from .core.atlas_glossary import AtlasGlossary
from .core.atlas_glossary_category import AtlasGlossaryCategory
Expand Down
2 changes: 2 additions & 0 deletions pyatlan/model/assets/core/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
from .anomalo_check import AnomaloCheck
from .app import App
from .application import Application
from .application_field import ApplicationField
from .asset import Asset
from .atlas_glossary import AtlasGlossary
from .atlas_glossary_category import AtlasGlossaryCategory
Expand Down Expand Up @@ -147,6 +148,7 @@
Stakeholder.Attributes.update_forward_refs(**localns)
AirflowDag.Attributes.update_forward_refs(**localns)
AirflowTask.Attributes.update_forward_refs(**localns)
ApplicationField.Attributes.update_forward_refs(**localns)
Application.Attributes.update_forward_refs(**localns)
AdfDataflow.Attributes.update_forward_refs(**localns)
AdfDataset.Attributes.update_forward_refs(**localns)
Expand Down
27 changes: 27 additions & 0 deletions pyatlan/model/assets/core/application.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,9 +54,16 @@ def __setattr__(self, name, value):
"""
TBC
"""
APPLICATION_CHILD_FIELDS: ClassVar[RelationField] = RelationField(
"applicationChildFields"
)
"""
TBC
"""

_convenience_properties: ClassVar[List[str]] = [
"application_owned_assets",
"application_child_fields",
]

@property
Expand All @@ -73,10 +80,29 @@ def application_owned_assets(self, application_owned_assets: Optional[List[Asset
self.attributes = self.Attributes()
self.attributes.application_owned_assets = application_owned_assets

@property
def application_child_fields(self) -> Optional[List[ApplicationField]]:
return (
None
if self.attributes is None
else self.attributes.application_child_fields
)

@application_child_fields.setter
def application_child_fields(
self, application_child_fields: Optional[List[ApplicationField]]
):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.application_child_fields = application_child_fields

class Attributes(App.Attributes):
application_owned_assets: Optional[List[Asset]] = Field(
default=None, description=""
) # relationship
application_child_fields: Optional[List[ApplicationField]] = Field(
default=None, description=""
) # relationship

@classmethod
@init_guid
Expand Down Expand Up @@ -108,4 +134,5 @@ def creator(
)


from .application_field import ApplicationField # noqa
from .asset import Asset # noqa
165 changes: 165 additions & 0 deletions pyatlan/model/assets/core/application_field.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,165 @@
# SPDX-License-Identifier: Apache-2.0
# Copyright 2022 Atlan Pte. Ltd.


from __future__ import annotations

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 RelationField
from pyatlan.utils import init_guid, validate_required_fields

from .app import App


class ApplicationField(App):
"""Description"""

@overload
@classmethod
def creator(
cls,
*,
name: str,
application_qualified_name: str,
) -> ApplicationField: ...

@overload
@classmethod
def creator(
cls,
*,
name: str,
application_qualified_name: str,
connection_qualified_name: str,
) -> ApplicationField: ...

@classmethod
@init_guid
def creator(
cls,
*,
name: str,
application_qualified_name: str,
connection_qualified_name: Optional[str] = None,
) -> ApplicationField:
validate_required_fields(
["name", "application_qualified_name"], [name, application_qualified_name]
)
attributes = ApplicationField.Attributes.create(
name=name,
application_qualified_name=application_qualified_name,
connection_qualified_name=connection_qualified_name,
)
return cls(attributes=attributes)

type_name: str = Field(default="ApplicationField", allow_mutation=False)

@validator("type_name")
def validate_type_name(cls, v):
if v != "ApplicationField":
raise ValueError("must be ApplicationField")
return v

def __setattr__(self, name, value):
if name in ApplicationField._convenience_properties:
return object.__setattr__(self, name, value)
super().__setattr__(name, value)

APPLICATION_PARENT: ClassVar[RelationField] = RelationField("applicationParent")
"""
TBC
"""
APPLICATION_FIELD_OWNED_ASSETS: ClassVar[RelationField] = RelationField(
"applicationFieldOwnedAssets"
)
"""
TBC
"""

_convenience_properties: ClassVar[List[str]] = [
"application_parent",
"application_field_owned_assets",
]

@property
def application_parent(self) -> Optional[Application]:
return None if self.attributes is None else self.attributes.application_parent

@application_parent.setter
def application_parent(self, application_parent: Optional[Application]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.application_parent = application_parent

@property
def application_field_owned_assets(self) -> Optional[List[Asset]]:
return (
None
if self.attributes is None
else self.attributes.application_field_owned_assets
)

@application_field_owned_assets.setter
def application_field_owned_assets(
self, application_field_owned_assets: Optional[List[Asset]]
):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.application_field_owned_assets = application_field_owned_assets

class Attributes(App.Attributes):
application_parent: Optional[Application] = Field(
default=None, description=""
) # relationship
application_field_owned_assets: Optional[List[Asset]] = Field(
default=None, description=""
) # relationship

@classmethod
@init_guid
def create(
cls,
*,
name: str,
application_qualified_name: str,
connection_qualified_name: Optional[str] = None,
) -> ApplicationField.Attributes:
validate_required_fields(
["name", "application_qualified_name"],
[name, application_qualified_name],
)
if connection_qualified_name:
connector_name = AtlanConnectorType.get_connector_name(
connection_qualified_name
)
else:
connection_qn, connector_name = AtlanConnectorType.get_connector_name(
application_qualified_name, "application_qualified_name", 4
)

return ApplicationField.Attributes(
name=name,
qualified_name=f"{application_qualified_name}/{name}",
connection_qualified_name=connection_qualified_name or connection_qn,
connector_name=connector_name,
application_parent=Application.ref_by_qualified_name(
application_qualified_name
),
)

attributes: ApplicationField.Attributes = Field(
default_factory=lambda: ApplicationField.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 .application import Application # noqa
from .asset import Asset # noqa
Loading
Loading