# SPDX-License-Identifier: Apache-2.0
# Copyright 2025 Atlan Pte. Ltd.
from __future__ import annotations
from datetime import datetime
from typing import ClassVar, Dict, List, Optional
from pydantic.v1 import Field, validator
from pyatlan.model.fields.atlan_fields import (
BooleanField,
KeywordField,
KeywordTextField,
NumericField,
RelationField,
TextField,
)
from .database import Database
[docs]
class MongoDBDatabase(Database):
"""Description"""
type_name: str = Field(default="MongoDBDatabase", allow_mutation=False)
@validator("type_name")
def validate_type_name(cls, v):
if v != "MongoDBDatabase":
raise ValueError("must be MongoDBDatabase")
return v
def __setattr__(self, name, value):
if name in MongoDBDatabase._convenience_properties:
return object.__setattr__(self, name, value)
super().__setattr__(name, value)
MONGO_DB_DATABASE_COLLECTION_COUNT: ClassVar[NumericField] = NumericField(
"mongoDBDatabaseCollectionCount", "mongoDBDatabaseCollectionCount"
)
"""
Number of collections in the database.
"""
SCHEMA_COUNT: ClassVar[NumericField] = NumericField("schemaCount", "schemaCount")
"""
Number of schemas in this database.
"""
QUERY_COUNT: ClassVar[NumericField] = NumericField("queryCount", "queryCount")
"""
Number of times this asset has been queried.
"""
QUERY_USER_COUNT: ClassVar[NumericField] = NumericField(
"queryUserCount", "queryUserCount"
)
"""
Number of unique users who have queried this asset.
"""
QUERY_USER_MAP: ClassVar[KeywordField] = KeywordField(
"queryUserMap", "queryUserMap"
)
"""
Map of unique users who have queried this asset to the number of times they have queried it.
"""
QUERY_COUNT_UPDATED_AT: ClassVar[NumericField] = NumericField(
"queryCountUpdatedAt", "queryCountUpdatedAt"
)
"""
Time (epoch) at which the query count was last updated, in milliseconds.
"""
DATABASE_NAME: ClassVar[KeywordTextField] = KeywordTextField(
"databaseName", "databaseName.keyword", "databaseName"
)
"""
Simple name of the database in which this SQL asset exists, or empty if it does not exist within a database.
"""
DATABASE_QUALIFIED_NAME: ClassVar[KeywordField] = KeywordField(
"databaseQualifiedName", "databaseQualifiedName"
)
"""
Unique name of the database in which this SQL asset exists, or empty if it does not exist within a database.
"""
SCHEMA_NAME: ClassVar[KeywordTextField] = KeywordTextField(
"schemaName", "schemaName.keyword", "schemaName"
)
"""
Simple name of the schema in which this SQL asset exists, or empty if it does not exist within a schema.
"""
SCHEMA_QUALIFIED_NAME: ClassVar[KeywordField] = KeywordField(
"schemaQualifiedName", "schemaQualifiedName"
)
"""
Unique name of the schema in which this SQL asset exists, or empty if it does not exist within a schema.
"""
TABLE_NAME: ClassVar[KeywordTextField] = KeywordTextField(
"tableName", "tableName.keyword", "tableName"
)
"""
Simple name of the table in which this SQL asset exists, or empty if it does not exist within a table.
"""
TABLE_QUALIFIED_NAME: ClassVar[KeywordField] = KeywordField(
"tableQualifiedName", "tableQualifiedName"
)
"""
Unique name of the table in which this SQL asset exists, or empty if it does not exist within a table.
"""
VIEW_NAME: ClassVar[KeywordTextField] = KeywordTextField(
"viewName", "viewName.keyword", "viewName"
)
"""
Simple name of the view in which this SQL asset exists, or empty if it does not exist within a view.
"""
VIEW_QUALIFIED_NAME: ClassVar[KeywordField] = KeywordField(
"viewQualifiedName", "viewQualifiedName"
)
"""
Unique name of the view in which this SQL asset exists, or empty if it does not exist within a view.
"""
CALCULATION_VIEW_NAME: ClassVar[KeywordTextField] = KeywordTextField(
"calculationViewName", "calculationViewName.keyword", "calculationViewName"
)
"""
Simple name of the calculation view in which this SQL asset exists, or empty if it does not exist within a calculation view.
""" # noqa: E501
CALCULATION_VIEW_QUALIFIED_NAME: ClassVar[KeywordField] = KeywordField(
"calculationViewQualifiedName", "calculationViewQualifiedName"
)
"""
Unique name of the calculation view in which this SQL asset exists, or empty if it does not exist within a calculation view.
""" # noqa: E501
IS_PROFILED: ClassVar[BooleanField] = BooleanField("isProfiled", "isProfiled")
"""
Whether this asset has been profiled (true) or not (false).
"""
LAST_PROFILED_AT: ClassVar[NumericField] = NumericField(
"lastProfiledAt", "lastProfiledAt"
)
"""
Time (epoch) at which this asset was last profiled, in milliseconds.
"""
NO_SQL_SCHEMA_DEFINITION: ClassVar[TextField] = TextField(
"noSQLSchemaDefinition", "noSQLSchemaDefinition"
)
"""
Represents attributes for describing the key schema for the table and indexes.
"""
MONGO_DB_COLLECTIONS: ClassVar[RelationField] = RelationField("mongoDBCollections")
"""
TBC
"""
_convenience_properties: ClassVar[List[str]] = [
"mongo_d_b_database_collection_count",
"schema_count",
"query_count",
"query_user_count",
"query_user_map",
"query_count_updated_at",
"database_name",
"database_qualified_name",
"schema_name",
"schema_qualified_name",
"table_name",
"table_qualified_name",
"view_name",
"view_qualified_name",
"calculation_view_name",
"calculation_view_qualified_name",
"is_profiled",
"last_profiled_at",
"no_s_q_l_schema_definition",
"mongo_d_b_collections",
]
@property
def mongo_d_b_database_collection_count(self) -> Optional[int]:
return (
None
if self.attributes is None
else self.attributes.mongo_d_b_database_collection_count
)
@mongo_d_b_database_collection_count.setter
def mongo_d_b_database_collection_count(
self, mongo_d_b_database_collection_count: Optional[int]
):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.mongo_d_b_database_collection_count = (
mongo_d_b_database_collection_count
)
@property
def schema_count(self) -> Optional[int]:
return None if self.attributes is None else self.attributes.schema_count
@schema_count.setter
def schema_count(self, schema_count: Optional[int]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.schema_count = schema_count
@property
def query_count(self) -> Optional[int]:
return None if self.attributes is None else self.attributes.query_count
@query_count.setter
def query_count(self, query_count: Optional[int]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.query_count = query_count
@property
def query_user_count(self) -> Optional[int]:
return None if self.attributes is None else self.attributes.query_user_count
@query_user_count.setter
def query_user_count(self, query_user_count: Optional[int]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.query_user_count = query_user_count
@property
def query_user_map(self) -> Optional[Dict[str, int]]:
return None if self.attributes is None else self.attributes.query_user_map
@query_user_map.setter
def query_user_map(self, query_user_map: Optional[Dict[str, int]]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.query_user_map = query_user_map
@property
def query_count_updated_at(self) -> Optional[datetime]:
return (
None if self.attributes is None else self.attributes.query_count_updated_at
)
@query_count_updated_at.setter
def query_count_updated_at(self, query_count_updated_at: Optional[datetime]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.query_count_updated_at = query_count_updated_at
@property
def database_name(self) -> Optional[str]:
return None if self.attributes is None else self.attributes.database_name
@database_name.setter
def database_name(self, database_name: Optional[str]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.database_name = database_name
@property
def database_qualified_name(self) -> Optional[str]:
return (
None if self.attributes is None else self.attributes.database_qualified_name
)
@database_qualified_name.setter
def database_qualified_name(self, database_qualified_name: Optional[str]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.database_qualified_name = database_qualified_name
@property
def schema_name(self) -> Optional[str]:
return None if self.attributes is None else self.attributes.schema_name
@schema_name.setter
def schema_name(self, schema_name: Optional[str]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.schema_name = schema_name
@property
def schema_qualified_name(self) -> Optional[str]:
return (
None if self.attributes is None else self.attributes.schema_qualified_name
)
@schema_qualified_name.setter
def schema_qualified_name(self, schema_qualified_name: Optional[str]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.schema_qualified_name = schema_qualified_name
@property
def table_name(self) -> Optional[str]:
return None if self.attributes is None else self.attributes.table_name
@table_name.setter
def table_name(self, table_name: Optional[str]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.table_name = table_name
@property
def table_qualified_name(self) -> Optional[str]:
return None if self.attributes is None else self.attributes.table_qualified_name
@table_qualified_name.setter
def table_qualified_name(self, table_qualified_name: Optional[str]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.table_qualified_name = table_qualified_name
@property
def view_name(self) -> Optional[str]:
return None if self.attributes is None else self.attributes.view_name
@view_name.setter
def view_name(self, view_name: Optional[str]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.view_name = view_name
@property
def view_qualified_name(self) -> Optional[str]:
return None if self.attributes is None else self.attributes.view_qualified_name
@view_qualified_name.setter
def view_qualified_name(self, view_qualified_name: Optional[str]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.view_qualified_name = view_qualified_name
@property
def calculation_view_name(self) -> Optional[str]:
return (
None if self.attributes is None else self.attributes.calculation_view_name
)
@calculation_view_name.setter
def calculation_view_name(self, calculation_view_name: Optional[str]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.calculation_view_name = calculation_view_name
@property
def calculation_view_qualified_name(self) -> Optional[str]:
return (
None
if self.attributes is None
else self.attributes.calculation_view_qualified_name
)
@calculation_view_qualified_name.setter
def calculation_view_qualified_name(
self, calculation_view_qualified_name: Optional[str]
):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.calculation_view_qualified_name = (
calculation_view_qualified_name
)
@property
def is_profiled(self) -> Optional[bool]:
return None if self.attributes is None else self.attributes.is_profiled
@is_profiled.setter
def is_profiled(self, is_profiled: Optional[bool]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.is_profiled = is_profiled
@property
def last_profiled_at(self) -> Optional[datetime]:
return None if self.attributes is None else self.attributes.last_profiled_at
@last_profiled_at.setter
def last_profiled_at(self, last_profiled_at: Optional[datetime]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.last_profiled_at = last_profiled_at
@property
def no_s_q_l_schema_definition(self) -> Optional[str]:
return (
None
if self.attributes is None
else self.attributes.no_s_q_l_schema_definition
)
@no_s_q_l_schema_definition.setter
def no_s_q_l_schema_definition(self, no_s_q_l_schema_definition: Optional[str]):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.no_s_q_l_schema_definition = no_s_q_l_schema_definition
@property
def mongo_d_b_collections(self) -> Optional[List[MongoDBCollection]]:
return (
None if self.attributes is None else self.attributes.mongo_d_b_collections
)
@mongo_d_b_collections.setter
def mongo_d_b_collections(
self, mongo_d_b_collections: Optional[List[MongoDBCollection]]
):
if self.attributes is None:
self.attributes = self.Attributes()
self.attributes.mongo_d_b_collections = mongo_d_b_collections
class Attributes(Database.Attributes):
mongo_d_b_database_collection_count: Optional[int] = Field(
default=None, description=""
)
schema_count: Optional[int] = Field(default=None, description="")
query_count: Optional[int] = Field(default=None, description="")
query_user_count: Optional[int] = Field(default=None, description="")
query_user_map: Optional[Dict[str, int]] = Field(default=None, description="")
query_count_updated_at: Optional[datetime] = Field(default=None, description="")
database_name: Optional[str] = Field(default=None, description="")
database_qualified_name: Optional[str] = Field(default=None, description="")
schema_name: Optional[str] = Field(default=None, description="")
schema_qualified_name: Optional[str] = Field(default=None, description="")
table_name: Optional[str] = Field(default=None, description="")
table_qualified_name: Optional[str] = Field(default=None, description="")
view_name: Optional[str] = Field(default=None, description="")
view_qualified_name: Optional[str] = Field(default=None, description="")
calculation_view_name: Optional[str] = Field(default=None, description="")
calculation_view_qualified_name: Optional[str] = Field(
default=None, description=""
)
is_profiled: Optional[bool] = Field(default=None, description="")
last_profiled_at: Optional[datetime] = Field(default=None, description="")
no_s_q_l_schema_definition: Optional[str] = Field(default=None, description="")
mongo_d_b_collections: Optional[List[MongoDBCollection]] = Field(
default=None, description=""
) # relationship
attributes: MongoDBDatabase.Attributes = Field(
default_factory=lambda: MongoDBDatabase.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 .mongo_d_b_collection import MongoDBCollection # noqa: E402, F401