from typing import Optional
from tdw_catalog import Catalog
from tdw_catalog.entity import Entity, Property, EntityBase
from tdw_catalog.errors import CatalogNotFoundException, _convert_error, _raise_error
[docs]@Entity([
Property("name", str),
Property("display_name", str),
Property("type", str, display_key="warehouse_type"),
Property("external", Optional[bool]),
])
class Warehouse(EntityBase):
"""
A :class:`.Warehouse` is a place where :class:`.Datasets` are stored.
Currently, :class:`.Warehouse`\\ s are configured at the deployment-level
and cannot be modified through this SDK.
Attributes
----------
name: str
The unique name of the warehouse in the system. This name will never
change for the life of the :class:`.Warehouse`.
display_name: str
The descriptive name of the :class:`.Warehouse`.
warehouse_type: str
The type of Warehouse this represents.
external: Optional[bool]
True if this Warehouse is virtualized within the :class:`.Catalog`
"""
_client: 'Catalog'
name: str
display_name: str
warehouse_type: str
external: Optional[bool]
[docs] @classmethod
def get(cls, client: 'Catalog', name: str, organization_id: str):
"""
Retrieve a Warehouse by name
Parameters
----------
client : Catalog
The :class:`.Catalog` client to use to get the :class:`.Warehouse`
name : str
The unique name of the :class:`.Warehouse`
organization_id : str
The organization in which the :class:`.Warehouse` is configured
Returns
-------
Warehouse
The :class:`.Warehouse` associated with the given name
Raises
------
CatalogPermissionDeniedException
If the caller is not allowed to retrieve the given :class:`.Warehouse`
CatalogNotFoundException
If the given :class:`.Warehouse` does not exist
CatalogException
If call to the :class:`.Catalog` server fails
"""
try:
res = client._list_warehouses(organization_id=organization_id)
w = next(filter(lambda w: w["name"] == name, res), None)
if w is None:
raise CatalogNotFoundException(
'Cannot find warehouse with name {name}'.format(name=name))
if "external" in w:
return ExternalWarehouse(client, **w)
else:
return TargetWarehouse(client, **w)
except Exception as e:
err = _raise_error(e, "Unable to fetch Warehouse {}".format(name))
def __repr__(self) -> str:
return self.__str__()
def __str__(self) -> str:
return f"<Warehouse name={self.name} display={self.display_name}>"
def __hash__(self) -> int:
return self.__str__().__hash__()
def __eq__(self, other) -> bool:
return isinstance(other, self.__class__) and self.name == other.name
[docs]class TargetWarehouse(Warehouse):
"""
A :class:`.TargetWarehouse` is a :class:`.Warehouse` which is configured for
data ingestion. New data can be written to a :class:`.TargetWarehouse`.
"""
[docs]@Entity([
Property("database_name", Optional[str]),
Property("schema", Optional[str])
])
class ExternalWarehouse(Warehouse):
"""
An :class:`.ExternalWarehouse` is a :class:`.Warehouse` which is configured for
Data Virtualization. New data cannot be written to an :class:`.ExternalWarehouse`,
but virtualized :class:`.Dataset`\\ s can be created which read from it.
Attributes
----------
database_name: Optional[str]
If set, the database to virtualize tables and views from
schema: Optional[str]
If set, the schema to virtualize tables and views from
"""
schema: Optional[str]