DomainsDictionary

A DomainsDictionary is used to map an entire database from an arbitrary schema to a PhenEx internal representation.

Attributes:

Name	Type	Description
domains_dict	Dict[str, class]	A dictionary where keys are domain names and values are uninstantiated PhenexTable class objects.

Methods:

Name	Description
get_mapped_tables	Get all tables mapped to PhenEx representation using the given connection.
set_mapped_tables	Create a view for all mapped tables in the destination database.

Source code in phenex/mappers.py

class DomainsDictionary:
    """
    A DomainsDictionary is used to map an entire database from an arbitrary schema to a PhenEx internal representation.

    Attributes:
        domains_dict (Dict[str, class]): A dictionary where keys are domain names and values are uninstantiated PhenexTable class objects.

    Methods:
        get_mapped_tables(con, database=None) -> Dict[str, Table]:
            Get all tables mapped to PhenEx representation using the given connection.
        set_mapped_tables(con, con, overwrite=False) -> None:
            Create a view for all mapped tables in the destination database.
    """

    def __init__(self, domains_dict):
        self.domains_dict = domains_dict

    def set_mapped_tables(self, con, overwrite=False) -> Dict[str, Table]:
        """
        Get all tables mapped to PhenEx representation using the given connection.

        If a database is not provided, the current database of the connection is used to find the tables.

        Args:
            con: The connection to the database.
            database (Optional[str]): The name of the database. Defaults to the current database of the connection.

        Returns:
            Dict[str, Table]: A dictionary where keys are domain names and values are mapped tables.
        """
        existing_tables = con.dest_connection.list_tables(
            database=con.SNOWFLAKE_DEST_DATABASE
        )
        for domain, mapper in self.domains_dict.items():
            if domain not in existing_tables or overwrite:
                t = con.get_source_table(mapper.NAME_TABLE)
                mapped_table = mapper(t).table
                con.create_view(
                    mapped_table, name_table=mapper.NAME_TABLE, overwrite=overwrite
                )

    def get_mapped_tables(self, con) -> Dict[str, PhenexTable]:
        """
        Get all tables mapped to PhenEx representation using the given connection.

        If a database is not provided, the current database of the connection is used to find the tables.

        Args:
            con: The connection to the database.
            database (Optional[str]): The name of the database. Defaults to the current database of the connection.

        Returns:
            Dict[str, PhenexTable]: A dictionary where keys are domain names and values are mapped tables.
        """
        # self.set_mapped_tables(con)
        mapped_tables = {}
        for domain, mapper in self.domains_dict.items():
            mapped_tables[domain] = mapper(con.get_source_table(mapper.NAME_TABLE))
        return mapped_tables

get_mapped_tables(con)

Get all tables mapped to PhenEx representation using the given connection.

If a database is not provided, the current database of the connection is used to find the tables.

Parameters:

Name	Type	Description	Default
con		The connection to the database.	required
database	Optional[str]	The name of the database. Defaults to the current database of the connection.	required

Returns:

Type	Description
Dict[str, PhenexTable]	Dict[str, PhenexTable]: A dictionary where keys are domain names and values are mapped tables.

Source code in phenex/mappers.py

def get_mapped_tables(self, con) -> Dict[str, PhenexTable]:
    """
    Get all tables mapped to PhenEx representation using the given connection.

    If a database is not provided, the current database of the connection is used to find the tables.

    Args:
        con: The connection to the database.
        database (Optional[str]): The name of the database. Defaults to the current database of the connection.

    Returns:
        Dict[str, PhenexTable]: A dictionary where keys are domain names and values are mapped tables.
    """
    # self.set_mapped_tables(con)
    mapped_tables = {}
    for domain, mapper in self.domains_dict.items():
        mapped_tables[domain] = mapper(con.get_source_table(mapper.NAME_TABLE))
    return mapped_tables

set_mapped_tables(con, overwrite=False)

Get all tables mapped to PhenEx representation using the given connection.

If a database is not provided, the current database of the connection is used to find the tables.

Parameters:

Name	Type	Description	Default
con		The connection to the database.	required
database	Optional[str]	The name of the database. Defaults to the current database of the connection.	required

Returns:

Type	Description
Dict[str, Table]	Dict[str, Table]: A dictionary where keys are domain names and values are mapped tables.

Source code in phenex/mappers.py

def set_mapped_tables(self, con, overwrite=False) -> Dict[str, Table]:
    """
    Get all tables mapped to PhenEx representation using the given connection.

    If a database is not provided, the current database of the connection is used to find the tables.

    Args:
        con: The connection to the database.
        database (Optional[str]): The name of the database. Defaults to the current database of the connection.

    Returns:
        Dict[str, Table]: A dictionary where keys are domain names and values are mapped tables.
    """
    existing_tables = con.dest_connection.list_tables(
        database=con.SNOWFLAKE_DEST_DATABASE
    )
    for domain, mapper in self.domains_dict.items():
        if domain not in existing_tables or overwrite:
            t = con.get_source_table(mapper.NAME_TABLE)
            mapped_table = mapper(t).table
            con.create_view(
                mapped_table, name_table=mapper.NAME_TABLE, overwrite=overwrite
            )