Source code for pyatlan.model.assets.cassandra_index

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


from __future__ import annotations

from typing import ClassVar, Dict, List, Optional

from pydantic.v1 import Field, validator

from pyatlan.model.fields.atlan_fields import KeywordField, RelationField

from .cassandra import Cassandra


[docs] class CassandraIndex(Cassandra): """Description""" type_name: str = Field(default="CassandraIndex", allow_mutation=False) @validator("type_name") def validate_type_name(cls, v): if v != "CassandraIndex": raise ValueError("must be CassandraIndex") return v def __setattr__(self, name, value): if name in CassandraIndex._convenience_properties: return object.__setattr__(self, name, value) super().__setattr__(name, value) CASSANDRA_INDEX_KIND: ClassVar[KeywordField] = KeywordField( "cassandraIndexKind", "cassandraIndexKind" ) """ Kind of index (e.g. COMPOSITES). """ CASSANDRA_INDEX_OPTIONS: ClassVar[KeywordField] = KeywordField( "cassandraIndexOptions", "cassandraIndexOptions" ) """ Options for the index. """ CASSANDRA_INDEX_QUERY: ClassVar[KeywordField] = KeywordField( "cassandraIndexQuery", "cassandraIndexQuery" ) """ Query used to create the index. """ CASSANDRA_TABLE: ClassVar[RelationField] = RelationField("cassandraTable") """ TBC """ _convenience_properties: ClassVar[List[str]] = [ "cassandra_index_kind", "cassandra_index_options", "cassandra_index_query", "cassandra_table", ] @property def cassandra_index_kind(self) -> Optional[str]: return None if self.attributes is None else self.attributes.cassandra_index_kind @cassandra_index_kind.setter def cassandra_index_kind(self, cassandra_index_kind: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.cassandra_index_kind = cassandra_index_kind @property def cassandra_index_options(self) -> Optional[Dict[str, str]]: return ( None if self.attributes is None else self.attributes.cassandra_index_options ) @cassandra_index_options.setter def cassandra_index_options( self, cassandra_index_options: Optional[Dict[str, str]] ): if self.attributes is None: self.attributes = self.Attributes() self.attributes.cassandra_index_options = cassandra_index_options @property def cassandra_index_query(self) -> Optional[str]: return ( None if self.attributes is None else self.attributes.cassandra_index_query ) @cassandra_index_query.setter def cassandra_index_query(self, cassandra_index_query: Optional[str]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.cassandra_index_query = cassandra_index_query @property def cassandra_table(self) -> Optional[CassandraTable]: return None if self.attributes is None else self.attributes.cassandra_table @cassandra_table.setter def cassandra_table(self, cassandra_table: Optional[CassandraTable]): if self.attributes is None: self.attributes = self.Attributes() self.attributes.cassandra_table = cassandra_table class Attributes(Cassandra.Attributes): cassandra_index_kind: Optional[str] = Field(default=None, description="") cassandra_index_options: Optional[Dict[str, str]] = Field( default=None, description="" ) cassandra_index_query: Optional[str] = Field(default=None, description="") cassandra_table: Optional[CassandraTable] = Field( default=None, description="" ) # relationship attributes: CassandraIndex.Attributes = Field( default_factory=lambda: CassandraIndex.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 .cassandra_table import CassandraTable # noqa: E402, F401 CassandraIndex.Attributes.update_forward_refs()