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 
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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.

        Raises:
            ValueError: If a required table is not found in the database.
        """
        # self.set_mapped_tables(con)
        mapped_tables = {}
        for domain, mapper in self.domains_dict.items():
            try:
                source_table = con.get_source_table(mapper.NAME_TABLE)
                mapped_tables[domain] = mapper(source_table)
            except exc.IbisError as e:
                if "Table not found" in str(e):
                    logger.warning(
                        f"Required table '{mapper.NAME_TABLE}' for domain '{domain}' not found in the database@ adding None for this domain"
                    )
                    mapped_tables[domain] = None
                else:
                    raise
        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

    def to_dict(self) -> dict:
        """
        Serialize the DomainsDictionary configuration.

        This serializes the mapping of domain names to PhenexTable classes,
        storing only the class configuration, not any actual table data.

        Returns:
            dict: Serialized domains dictionary configuration
        """
        serialized_domains = {}
        for domain_name, mapper_class in self.domains_dict.items():
            # Store the class configuration
            serialized_domains[domain_name] = mapper_class.to_dict()

        return {
            "class_name": "DomainsDictionary",
            "domains_dict": serialized_domains,
        }

    @staticmethod
    def from_dict(data: dict) -> "DomainsDictionary":
        """
        Reconstruct a DomainsDictionary from serialized data.

        Args:
            data: Serialized DomainsDictionary configuration

        Returns:
            DomainsDictionary instance with mapper classes
        """
        from phenex.tables import PhenexTable
        import importlib

        domains_dict = {}
        for domain_name, table_config in data.get("domains_dict", {}).items():
            # Reconstruct the mapper class by looking it up in globals
            table_class_name = table_config["__table_class__"]

            # Try to find the class in the current module (mappers.py)
            if table_class_name in globals():
                mapper_class = globals()[table_class_name]
            else:
                # Try to import from the module specified
                module_name = table_config.get("__module__", "phenex.mappers")
                try:
                    module = importlib.import_module(module_name)
                    mapper_class = getattr(module, table_class_name)
                except (ImportError, AttributeError):
                    raise ValueError(
                        f"Cannot find mapper class '{table_class_name}' in module '{module_name}'"
                    )

            domains_dict[domain_name] = mapper_class

        return DomainsDictionary(domains_dict)

from_dict(data) staticmethod

Reconstruct a DomainsDictionary from serialized data.

Parameters:

Name Type Description Default
data dict

Serialized DomainsDictionary configuration

 required

Returns:

Type Description
DomainsDictionary

DomainsDictionary instance with mapper classes
Source code in phenex/mappers.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@staticmethod
def from_dict(data: dict) -> "DomainsDictionary":
    """
    Reconstruct a DomainsDictionary from serialized data.

    Args:
        data: Serialized DomainsDictionary configuration

    Returns:
        DomainsDictionary instance with mapper classes
    """
    from phenex.tables import PhenexTable
    import importlib

    domains_dict = {}
    for domain_name, table_config in data.get("domains_dict", {}).items():
        # Reconstruct the mapper class by looking it up in globals
        table_class_name = table_config["__table_class__"]

        # Try to find the class in the current module (mappers.py)
        if table_class_name in globals():
            mapper_class = globals()[table_class_name]
        else:
            # Try to import from the module specified
            module_name = table_config.get("__module__", "phenex.mappers")
            try:
                module = importlib.import_module(module_name)
                mapper_class = getattr(module, table_class_name)
            except (ImportError, AttributeError):
                raise ValueError(
                    f"Cannot find mapper class '{table_class_name}' in module '{module_name}'"
                )

        domains_dict[domain_name] = mapper_class

    return DomainsDictionary(domains_dict)

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.

Raises:

Type Description
ValueError

If a required table is not found in the database.
Source code in phenex/mappers.py 
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
83
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.

    Raises:
        ValueError: If a required table is not found in the database.
    """
    # self.set_mapped_tables(con)
    mapped_tables = {}
    for domain, mapper in self.domains_dict.items():
        try:
            source_table = con.get_source_table(mapper.NAME_TABLE)
            mapped_tables[domain] = mapper(source_table)
        except exc.IbisError as e:
            if "Table not found" in str(e):
                logger.warning(
                    f"Required table '{mapper.NAME_TABLE}' for domain '{domain}' not found in the database@ adding None for this domain"
                )
                mapped_tables[domain] = None
            else:
                raise
    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 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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
            )

to_dict()

Serialize the DomainsDictionary configuration.

This serializes the mapping of domain names to PhenexTable classes, storing only the class configuration, not any actual table data.

Returns:

Name Type Description
dict dict

Serialized domains dictionary configuration
Source code in phenex/mappers.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def to_dict(self) -> dict:
    """
    Serialize the DomainsDictionary configuration.

    This serializes the mapping of domain names to PhenexTable classes,
    storing only the class configuration, not any actual table data.

    Returns:
        dict: Serialized domains dictionary configuration
    """
    serialized_domains = {}
    for domain_name, mapper_class in self.domains_dict.items():
        # Store the class configuration
        serialized_domains[domain_name] = mapper_class.to_dict()

    return {
        "class_name": "DomainsDictionary",
        "domains_dict": serialized_domains,
    }