Source code for deephaven_enterprise.database

#  Copyright (c) 2016-2023 Deephaven Data Labs and Patent Pending

import jpy
from deephaven import DHError
from deephaven.table import Table, PartitionedTable
from deephaven.jcompat import wrap_j_object
from typing import Any, List

class Database:
    _j_object_type = jpy.get_type("io.deephaven.enterprise.database.Database")

    def __init__(self, j_db: jpy.JType):
        self.j_db = jpy.cast(j_db, self._j_object_type)
        if self.j_db is None:
            raise DHError("j_db type is not io.deephaven.enterprise.database.Database")

    def _j_collection_to_list(self, jcollection) -> List[Any]:
        """Converts a java Collection to a python list."""
        result = []
        if not jcollection:
            return result

        it = jcollection.iterator()
        while it.hasNext():
            result.append(wrap_j_object(it.next()))

        return result

    def live_table(self, namespace: str, table_name: str, is_refreshing=True):
        """
        Fetch a live Table for the specified namespace and table name.

        :param namespace: the namespace
        :param table_name: the table
        :param is_refreshing: True if the returned table should be refreshing (defaults to True)
        :return: the live table at the specified namespace and table name
        """
        return Table(j_table=self.j_db.liveTable(namespace, table_name, is_refreshing))

    def live_partitioned_table(self, namespace: str, table_name: str):
        """
        Retrieve the specified historical table as a partitioned table from the Database.

        :param namespace: the namespace
        :param table_name: the table
        :return: a new historical partitioned table at the specified namespace and table name
        """
        return PartitionedTable(j_partitioned_table=self.j_db.livePartitionedTable(namespace, table_name))

    def historical_table(self, namespace: str, table_name: str):
        """
        Fetch a historical Table for the specified namespace and table name.

        :param namespace: the namespace
        :param table_name: the table
        :return: the historical table at the specified namespace and table name
        """
        return Table(j_table=self.j_db.historicalTable(namespace, table_name))

    def historical_partitioned_table(self, namespace: str, table_name: str):
        """
        Retrieve the specified historical table as a partitioned table from the Database.

        :param namespace: the namespace
        :param table_name: the table
        :return: a new historical partitioned table at the specified namespace and table name
        """
        return PartitionedTable(j_partitioned_table=self.j_db.historicalPartitionedTable(namespace, table_name))

    def namespaces(self):
        """
        Return the list of namespaces.

        :return: the list of namespaces
        """
        return self._j_collection_to_list(self.j_db.getNamespaces())

    def table_names(self, namespace: str):
        """
        Return the list tables within a namespace.

        :param namespace: the namespace

        :return: the list of tables within namespace
        """
        return self._j_collection_to_list(self.j_db.getTableNames(namespace))

    def catalog_table(self):
        """
        Return a table of the available tables.

        The result table contains a column for Namespace, TableName, and NamespaceSet.

        :return: a table of table names
        """
        return Table(j_table=self.j_db.getCatalogTable())

    def table_definition(self, namespace: str, table_name: str):
        """
        Fetch a Table containing the column definitions for the specified namespace and table name.

        :param namespace: the namespace
        :param table_name: the table
        :return: the definition table for the specified namespace and table name
        """
        return Table(j_table=self.j_db.getTableDefinitionTable(namespace, table_name))

db: Database = None