Skip to content

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.
get_source_tables

Get all source tables using the given connection.
set_mapped_tables

Create a view for all mapped tables in the destination database.
Source code in phenex/mappers.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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) -> Dict[str, Table]:
            Get all tables mapped to PhenEx representation using the given connection.
        get_source_tables(con) -> Dict[str, Table]:
            Get all source tables using the given connection.
        set_mapped_tables(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]:
        """
        Create a view for all mapped tables in the destination database.

        Args:
            con: The connection to the database.
            overwrite: Whether to overwrite existing views if found. Otherwise, throws an error.

        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:
                t = con.get_source_table(mapper.NAME_TABLE)
                mapped_table = mapper(t).table
                # overwrite error handling handled in create_view call
                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.

        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

    def get_source_tables(self, con) -> Dict[str, str]:
        """
        Get all source tables using the given connection.

        Args:
            con: The connection to the database.

        Returns:
            Dict[str, str]: A dictionary where keys are the source table names and values are table names.
        """
        source_tables = {}
        for mapper in self.domains_dict.values():
            table_name = mapper.NAME_TABLE
            if table_name not in source_tables:
                source_tables[table_name] = con.get_source_table(table_name)
        return source_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

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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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.

    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_source_tables(con)

Get all source tables using the given connection.

Parameters:

Name Type Description Default
con

The connection to the database.

 required

Returns:

Type Description
Dict[str, str]

Dict[str, str]: A dictionary where keys are the source table names and values are table names.
Source code in phenex/mappers.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get_source_tables(self, con) -> Dict[str, str]:
    """
    Get all source tables using the given connection.

    Args:
        con: The connection to the database.

    Returns:
        Dict[str, str]: A dictionary where keys are the source table names and values are table names.
    """
    source_tables = {}
    for mapper in self.domains_dict.values():
        table_name = mapper.NAME_TABLE
        if table_name not in source_tables:
            source_tables[table_name] = con.get_source_table(table_name)
    return source_tables

set_mapped_tables(con, overwrite=False)

Create a view for all mapped tables in the destination database.

Parameters:

Name Type Description Default
con

The connection to the database.

 required
overwrite

Whether to overwrite existing views if found. Otherwise, throws an error.

 False

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 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def set_mapped_tables(self, con, overwrite=False) -> Dict[str, Table]:
    """
    Create a view for all mapped tables in the destination database.

    Args:
        con: The connection to the database.
        overwrite: Whether to overwrite existing views if found. Otherwise, throws an error.

    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:
            t = con.get_source_table(mapper.NAME_TABLE)
            mapped_table = mapper(t).table
            # overwrite error handling handled in create_view call
            con.create_view(
                mapped_table, name_table=mapper.NAME_TABLE, overwrite=overwrite
            )