Source code for pyatlan.model.assets.a_d_l_s_container

# SPDX-License-Identifier: Apache-2.0
# Copyright 2025 Atlan Pte. Ltd.


from __future__ import annotations

from typing import ClassVar, List, Optional, overload
from warnings import warn

from pydantic.v1 import Field, validator

from pyatlan.model.enums import ADLSLeaseState, ADLSLeaseStatus, AtlanConnectorType
from pyatlan.model.fields.atlan_fields import (
    BooleanField,
    KeywordField,
    KeywordTextField,
    NumericField,
    RelationField,
)
from pyatlan.utils import init_guid, validate_required_fields

from .a_d_l_s import ADLS


[docs] class ADLSContainer(ADLS): """Description""" @overload @classmethod def creator( cls, *, name: str, adls_account_qualified_name: str, ) -> ADLSContainer: ... @overload @classmethod def creator( cls, *, name: str, adls_account_qualified_name: str, connection_qualified_name: str, ) -> ADLSContainer: ... @classmethod @init_guid def creator( cls, *, name: str, adls_account_qualified_name: str, connection_qualified_name: Optional[str] = None, ) -> ADLSContainer: validate_required_fields( ["name", "adls_account_qualified_name"], [name, adls_account_qualified_name] ) attributes = ADLSContainer.Attributes.create( name=name, adls_account_qualified_name=adls_account_qualified_name, connection_qualified_name=connection_qualified_name, ) return cls(attributes=attributes) @classmethod @init_guid def create(cls, *, name: str, adls_account_qualified_name: str) -> ADLSContainer: warn( ( "This method is deprecated, please use 'creator' " "instead, which offers identical functionality." ), DeprecationWarning, stacklevel=2, ) return cls.creator( name=name, adls_account_qualified_name=adls_account_qualified_name ) type_name: str = Field(default="ADLSContainer", allow_mutation=False) @validator("type_name") def validate_type_name(cls, v): if v != "ADLSContainer": raise ValueError("must be ADLSContainer") return v def __setattr__(self, name, value): if name in ADLSContainer._convenience_properties: return object.__setattr__(self, name, value) super().__setattr__(name, value) ADLS_CONTAINER_URL: ClassVar[KeywordTextField] = KeywordTextField( "adlsContainerUrl", "adlsContainerUrl.keyword", "adlsContainerUrl" ) """ URL of this container. """ ADLS_CONTAINER_LEASE_STATE: ClassVar[KeywordField] = KeywordField( "adlsContainerLeaseState", "adlsContainerLeaseState" ) """ Lease state of this container. """ ADLS_CONTAINER_LEASE_STATUS: ClassVar[KeywordField] = KeywordField( "adlsContainerLeaseStatus", "adlsContainerLeaseStatus" ) """ Lease status of this container. """ ADLS_CONTAINER_ENCRYPTION_SCOPE: ClassVar[KeywordField] = KeywordField( "adlsContainerEncryptionScope", "adlsContainerEncryptionScope" ) """ Encryption scope of this container. """ ADLS_CONTAINER_VERSION_LEVEL_IMMUTABILITY_SUPPORT: ClassVar[BooleanField] = ( BooleanField( "adlsContainerVersionLevelImmutabilitySupport", "adlsContainerVersionLevelImmutabilitySupport", ) ) """ Whether this container supports version-level immutability (true) or not (false). """ ADLS_OBJECT_COUNT: ClassVar[NumericField] = NumericField( "adlsObjectCount", "adlsObjectCount" ) """ Number of objects that exist within this container. """ ADLS_ACCOUNT: ClassVar[RelationField] = RelationField("adlsAccount") """ TBC """ ADLS_OBJECTS: ClassVar[RelationField] = RelationField("adlsObjects") """ TBC """ _convenience_properties: ClassVar[List[str]] = [ "adls_container_url", "adls_container_lease_state", "adls_container_lease_status", "adls_container_encryption_scope", "adls_container_version_level_immutability_support", "adls_object_count", "adls_account", "adls_objects", ] @property def adls_container_url(self) -> Optional[str]: return None if self.attributes is None else self.attributes.adls_container_url @adls_container_url.setter def adls_container_url(self, adls_container_url: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.adls_container_url = adls_container_url @property def adls_container_lease_state(self) -> Optional[ADLSLeaseState]: return ( None if self.attributes is None else self.attributes.adls_container_lease_state ) @adls_container_lease_state.setter def adls_container_lease_state( self, adls_container_lease_state: Optional[ADLSLeaseState] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.adls_container_lease_state = adls_container_lease_state @property def adls_container_lease_status(self) -> Optional[ADLSLeaseStatus]: return ( None if self.attributes is None else self.attributes.adls_container_lease_status ) @adls_container_lease_status.setter def adls_container_lease_status( self, adls_container_lease_status: Optional[ADLSLeaseStatus] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.adls_container_lease_status = adls_container_lease_status @property def adls_container_encryption_scope(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.adls_container_encryption_scope ) @adls_container_encryption_scope.setter def adls_container_encryption_scope( self, adls_container_encryption_scope: Optional[str] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.adls_container_encryption_scope = ( adls_container_encryption_scope ) @property def adls_container_version_level_immutability_support(self) -> Optional[bool]: return ( None if self.attributes is None else self.attributes.adls_container_version_level_immutability_support ) @adls_container_version_level_immutability_support.setter def adls_container_version_level_immutability_support( self, adls_container_version_level_immutability_support: Optional[bool] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.adls_container_version_level_immutability_support = ( adls_container_version_level_immutability_support ) @property def adls_object_count(self) -> Optional[int]: return None if self.attributes is None else self.attributes.adls_object_count @adls_object_count.setter def adls_object_count(self, adls_object_count: Optional[int]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.adls_object_count = adls_object_count @property def adls_account(self) -> Optional[ADLSAccount]: return None if self.attributes is None else self.attributes.adls_account @adls_account.setter def adls_account(self, adls_account: Optional[ADLSAccount]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.adls_account = adls_account @property def adls_objects(self) -> Optional[List[ADLSObject]]: return None if self.attributes is None else self.attributes.adls_objects @adls_objects.setter def adls_objects(self, adls_objects: Optional[List[ADLSObject]]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.adls_objects = adls_objects class Attributes(ADLS.Attributes): adls_container_url: Optional[str] = Field(default=None, description="") adls_container_lease_state: Optional[ADLSLeaseState] = Field( default=None, description="" ) adls_container_lease_status: Optional[ADLSLeaseStatus] = Field( default=None, description="" ) adls_container_encryption_scope: Optional[str] = Field( default=None, description="" ) adls_container_version_level_immutability_support: Optional[bool] = Field( default=None, description="" ) adls_object_count: Optional[int] = Field(default=None, description="") adls_account: Optional[ADLSAccount] = Field( default=None, description="" ) # relationship adls_objects: Optional[List[ADLSObject]] = Field( default=None, description="" ) # relationship @classmethod @init_guid def create( cls, *, name: str, adls_account_qualified_name: str, connection_qualified_name: Optional[str] = None, ) -> ADLSContainer.Attributes: validate_required_fields( ["name", "adls_account_qualified_name"], [name, adls_account_qualified_name], ) if connection_qualified_name: connector_name = AtlanConnectorType.get_connector_name( connection_qualified_name ) else: connection_qn, connector_name = AtlanConnectorType.get_connector_name( adls_account_qualified_name, "adls_account_qualified_name", 4 ) return ADLSContainer.Attributes( name=name, qualified_name=f"{adls_account_qualified_name}/{name}", adls_account=ADLSAccount.ref_by_qualified_name( adls_account_qualified_name ), adls_account_qualified_name=adls_account_qualified_name, adls_account_name=adls_account_qualified_name.split("/")[-1], connector_name=connector_name, connection_qualified_name=connection_qualified_name or connection_qn, ) attributes: ADLSContainer.Attributes = Field( default_factory=lambda: ADLSContainer.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 .a_d_l_s_account import ADLSAccount # noqa: E402, F401 from .a_d_l_s_object import ADLSObject # noqa: E402, F401 ADLSContainer.Attributes.update_forward_refs()