Skip to content

DuckdbTablesBuilder

A class to build and manage schema tables in DuckDB.

This class extends SchemaBuilder and provides functionality to create metadata, dimension, and fact tables in DuckDB, while logging all operations.

Attributes:

Name Type Description
conn DuckDBPyConnection

DuckDB connection object.

Methods:

Name Description
build

Execute the full pipeline to create metadata, dimension tables, and a fact table.
build_duckdb_schema

Build the entire schema in DuckDB, including metadata, dimension, and fact tables.
create_dimension_tables

Generate dimension tables for categorical columns in the dataset.
create_duckdb_dimension_tables

Create dimension tables in DuckDB for categorical variables.
create_duckdb_fact_table

Create a fact table in DuckDB with foreign key relationships to dimension tables.
create_duckdb_metadata_table

Create a metadata table in DuckDB.
create_fact_table

Generate a fact table by replacing categorical values with corresponding IDs.
create_metadata_table

Automatically infer metadata for the DataFrame's columns, including types, labels,

display_schema

Display the structure of all tables in the DuckDB schema.
Source code in dashboard_template_database/builders/tables.py 
 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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
class DuckdbTablesBuilder(SchemaBuilder):
    """
    A class to build and manage schema tables in DuckDB.

    This class extends `SchemaBuilder` and provides functionality to create 
    metadata, dimension, and fact tables in DuckDB, while logging all operations.

    Attributes:
        conn (duckdb.DuckDBPyConnection): DuckDB connection object.
    """

    # Initialisation
    def __init__(self, df: pd.DataFrame, categorical_threshold: Optional[int] = 50, connection: Optional[duckdb.DuckDBPyConnection] = None, path : Optional[Union[os.PathLike, None]]=None, log_filename: Optional[os.PathLike] = os.path.join(FILE_PATH.parents[2], "logs/duckdb_schema_builder.log")):
        """
        Initialize the DuckdbTablesBuilder class.

        Args:
            df (pd.DataFrame): The input dataset.
            categorical_threshold (Optional[int]): Threshold for determining categorical variables.
            connection (Optional[duckdb.DuckDBPyConnection]): Existing DuckDB connection. Defaults to None.
            path (Optional[Union[os.PathLike, None]]): Path to the DuckDB database file. Defaults to None.
            log_filename (Optional[os.PathLike]): Path to the log file. Defaults to a pre-defined path.

        """
        # Initialisation du schéma
        super().__init__(df=df, categorical_threshold=categorical_threshold, log_filename=log_filename)

        # Initialisation de la connection
        if (connection is None) & (path is None) :
            self.conn = duckdb.connect(':memory:')
        elif (connection is None) :
            self.conn = duckdb.connect(path)
        else :
            self.conn = connection

    # Méthode de création de la table des méta-données
    def create_duckdb_metadata_table(self, table_name: Optional[str] = 'metadata', column_labels: Optional[Dict[str, str]] = None) -> None:
        """
        Create a metadata table in DuckDB.

        Args:
            table_name (Optional[str]): Name of the metadata table in DuckDB. Defaults to 'metadata'.
            column_labels (Optional[Dict[str, str]]): Optional mapping of column names to labels.

        """
        # Création de la table des méta-données si elle n'existe pas déjà
        if not hasattr(self, 'df_metadata'):
            _ = self.create_metadata_table(column_labels)

        # Conversion DataFrame en table DuckDB
        self.conn.register('temp_metadata', self.df_metadata)
        self.conn.execute(f"""
            CREATE TABLE {table_name} AS 
            SELECT * FROM temp_metadata
        """)
        self.conn.execute('DROP VIEW temp_metadata')

        # Logging
        self.logger.info("Successfully registered duckdb meta-data table")

    # Méthode de création des tables de dimensions
    def create_duckdb_dimension_tables(self, table_prefix: Optional[str] = 'dim_', column_labels: Optional[Dict[str, str]] = None) -> None:
        """
        Create dimension tables in DuckDB for categorical variables.

        Args:
            table_prefix (Optional[str]): Prefix for dimension table names. Defaults to 'dim_'.
            column_labels (Optional[Dict[str, str]]): Optional mapping of column names to labels.

        """
        # Création du dictionnaire des tables de dimensions si elles n'existent pas déjà
        if not hasattr(self, 'dimension_tables'):
            _ = self.create_dimension_tables(column_labels)

        # Création de chaque table de dimension dans DuckDB
        for dim_name, dim_df in self.dimension_tables.items():
            # Initialisation du nom de la table
            table_name = f"{table_prefix}{dim_name}"
            # Enregistrement d'une vue temporaire
            self.conn.register('temp_dim', dim_df)

            # Création d'une table avec "value" comme clé primaire
            self.conn.execute(f"""
                CREATE TABLE {table_name} AS 
                SELECT
                    value,
                    label 
                FROM temp_dim
            """)

            # Ajout de la vue correspondante
            self.conn.execute('DROP VIEW temp_dim')

            # Logging
            self.logger.info(f"Successfully registered duckdb dimension table for {dim_name}")

    # Méthode de création de la table d'informations
    def create_duckdb_fact_table(self, table_name: Optional[str] = 'fact_table', table_prefix: Optional[str] = 'dim_', column_labels: Optional[Dict[str, str]] = None) -> None:
        """
        Create a fact table in DuckDB with foreign key relationships to dimension tables.

        Args:
            table_name (Optional[str]): Name of the fact table in DuckDB. Defaults to 'fact_table'.
            table_prefix (Optional[str]): Prefix for dimension table names. Defaults to 'dim_'.
            column_labels (Optional[Dict[str, str]]): Optional mapping of column names to labels.

        """
        # Création de la table d'informations si elle n'existe pas déjà
        if not hasattr(self, 'df_fact'):
            _ = self.create_fact_table(column_labels)

        # Enregistrement de la table comme vue temporaire
        self.conn.register('temp_fact', self.df_fact)

        # Initialisation de la liste des conditions de jointure avec les tables de dimensions
        join_conditions = []
        # Initialisation de la liste des colonnes à sélectionner
        # select_columns = []

        # Parcours des tables de dimensions correspondant à des clés étrangères dans la table d'informations
        for dim_name in self.dimension_tables.keys():
            # Initialisation du nom de la table
            dim_table = f"{table_prefix}{dim_name}"

            # Ajout des conditions de jointure sur la base de la colonne "value" des tables de dimensions
            join_conditions.append(
                f"LEFT JOIN {dim_table} ON temp_fact.{dim_name} = {dim_table}.value"
            )

            # Logging
            self.logger.info(f"Successfully created foreign key for dimension '{dim_name}'")
            # Ajout de la colonne qui correspond à une clé étrangère
            # select_columns.append(f"temp_fact.{dim_name}")

        # Ajout des colonnes des variables non catégorielles
        # select_columns.extend([f"temp_fact.{col}" for col in self.df_fact.columns if col not in self.dimension_tables.keys()])

        # Création de la table d'information
        query = f"""
        CREATE TABLE {table_name} AS 
        SELECT *
        FROM temp_fact
        {' '.join(join_conditions)}
        """

        self.conn.execute(query)

        # Ajout de la vue
        self.conn.execute('DROP VIEW temp_fact')

        # Logging
        self.logger.info(f"Successfully registered duckdb fact table")

    # Méthode de construction du schéma
    def build_duckdb_schema(self, metadata_table: Optional[str] = 'metadata', fact_table: Optional[str] = 'fact_table', dim_table_prefix: Optional[str] = 'dim_', column_labels: Optional[Dict[str, str]] = None) -> None:
        """
        Build the entire schema in DuckDB, including metadata, dimension, and fact tables.

        Args:
            metadata_table (Optional[str]): Name of the metadata table. Defaults to 'metadata'.
            fact_table (Optional[str]): Name of the fact table. Defaults to 'fact_table'.
            dim_table_prefix (Optional[str]): Prefix for dimension tables. Defaults to 'dim_'.
            column_labels (Optional[Dict[str, str]]): Optional mapping of column names to labels.

        """
        # Création de la table des méta-données
        self.create_duckdb_metadata_table(
            table_name=metadata_table, 
            column_labels=column_labels
        )

        # Création de la table de dimensions
        self.create_duckdb_dimension_tables(
            table_prefix=dim_table_prefix, 
            column_labels=column_labels
        )

        # Création de la table d'informations avec les clés étrangères
        self.create_duckdb_fact_table(
            table_name=fact_table, 
            table_prefix=dim_table_prefix, 
            column_labels=column_labels
        )

    # Méthode d'affichage du schéma
    def display_schema(self) -> None:
        """
        Display the structure of all tables in the DuckDB schema.
        """
        # Extraction des tables
        tables = self.conn.execute("SHOW TABLES").fetchall()
        # Logging
        self.logger.info("\n Created Tables:")
        # Parcours des tables
        for table in tables:
            # Affichage de la structure
            self.logger.info(f"\n {table[0]} Structure:")
            # Extraction des informations relatives à la table
            table_info = self.conn.execute(f"DESCRIBE {table[0]}").fetchall()
            # Affichage de chaque information
            for col in table_info:
                self.logger.info(f"  {col[0]}: {col[1]}")

build 

build(column_labels: Optional[Union[Dict[str, str], None]] = None) -> Tuple[DataFrame, Dict[str, DataFrame], DataFrame]

Execute the full pipeline to create metadata, dimension tables, and a fact table.

Parameters:

Name Type Description Default

column_labels

dict

A dictionary mapping column names to labels. Defaults to None.

 None

Returns:

Name Type Description
tuple Tuple[DataFrame, Dict[str, DataFrame], DataFrame]

A tuple containing: - Metadata DataFrame. - Dictionary of dimension tables. - Fact table DataFrame.
Source code in dashboard_template_database/builders/schema.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def build(self, column_labels : Optional[Union[Dict[str, str], None]]= None) -> Tuple[pd.DataFrame, Dict[str, pd.DataFrame], pd.DataFrame] :
    """
    Execute the full pipeline to create metadata, dimension tables, and a fact table.

    Args:
        column_labels (dict, optional): A dictionary mapping column names to labels.
                                        Defaults to None.

    Returns:
        tuple: A tuple containing:
               - Metadata DataFrame.
               - Dictionary of dimension tables.
               - Fact table DataFrame.
    """
    # Création de la table des méta-données
    _ = self.create_metadata_table(column_labels=column_labels)
    # Création des tables de dimension
    _ = self.create_dimension_tables(column_labels=column_labels)
    # Création des tables d'informations
    _ = self.create_fact_table(column_labels=column_labels)

    return self.df_metadata, self.dimension_tables, self.df_fact

build_duckdb_schema 

build_duckdb_schema(metadata_table: Optional[str] = 'metadata', fact_table: Optional[str] = 'fact_table', dim_table_prefix: Optional[str] = 'dim_', column_labels: Optional[Dict[str, str]] = None) -> None

Build the entire schema in DuckDB, including metadata, dimension, and fact tables.

Parameters:

Name Type Description Default

metadata_table

Optional[str]

Name of the metadata table. Defaults to 'metadata'.

 'metadata'

fact_table

Optional[str]

Name of the fact table. Defaults to 'fact_table'.

 'fact_table'

dim_table_prefix

Optional[str]

Prefix for dimension tables. Defaults to 'dim_'.

 'dim_'

column_labels

Optional[Dict[str, str]]

Optional mapping of column names to labels.

 None
Source code in dashboard_template_database/builders/tables.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def build_duckdb_schema(self, metadata_table: Optional[str] = 'metadata', fact_table: Optional[str] = 'fact_table', dim_table_prefix: Optional[str] = 'dim_', column_labels: Optional[Dict[str, str]] = None) -> None:
    """
    Build the entire schema in DuckDB, including metadata, dimension, and fact tables.

    Args:
        metadata_table (Optional[str]): Name of the metadata table. Defaults to 'metadata'.
        fact_table (Optional[str]): Name of the fact table. Defaults to 'fact_table'.
        dim_table_prefix (Optional[str]): Prefix for dimension tables. Defaults to 'dim_'.
        column_labels (Optional[Dict[str, str]]): Optional mapping of column names to labels.

    """
    # Création de la table des méta-données
    self.create_duckdb_metadata_table(
        table_name=metadata_table, 
        column_labels=column_labels
    )

    # Création de la table de dimensions
    self.create_duckdb_dimension_tables(
        table_prefix=dim_table_prefix, 
        column_labels=column_labels
    )

    # Création de la table d'informations avec les clés étrangères
    self.create_duckdb_fact_table(
        table_name=fact_table, 
        table_prefix=dim_table_prefix, 
        column_labels=column_labels
    )

create_dimension_tables 

create_dimension_tables(column_labels: Optional[Union[Dict[str, str], None]] = None) -> Dict[str, DataFrame]

Generate dimension tables for categorical columns in the dataset.

Parameters:

Name Type Description Default

column_labels

dict

A dictionary mapping column names to labels. Defaults to None.

 None

Returns:

Name Type Description
dict Dict[str, DataFrame]

A dictionary of DataFrames, where keys are column names and values are the corresponding dimension tables.
Source code in dashboard_template_database/builders/schema.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def create_dimension_tables(self, column_labels : Optional[Union[Dict[str, str], None]]= None) -> Dict[str, pd.DataFrame] :
    """
    Generate dimension tables for categorical columns in the dataset.

    Args:
        column_labels (dict, optional): A dictionary mapping column names to labels.
                                        Defaults to None.

    Returns:
        dict: A dictionary of DataFrames, where keys are column names and values are 
              the corresponding dimension tables.
    """
    # Création d'une table de méta-données si cette-dernière n'existe pas déjà
    if not hasattr(self, 'metadata') :
        _ = self.create_metadata_table(column_labels=column_labels)

    # Initialisation du dictionnaire des tables de dimension
    self.dimension_tables = {}

    # Parcours des tables de dimensions
    for categorical_dimension in self.df_metadata.loc[self.df_metadata['is_categorical'], 'name'] :
        # Extraction des modalités
        self.dimension_tables[categorical_dimension] = pd.Series(self.df[categorical_dimension].unique(), name='label').to_frame().reset_index(names='value').sort_values(by='label', ascending=True, ignore_index=True)
        # Logging
        self.logger.info(f"Successfully built dimension table for '{categorical_dimension}'")

    # Logging
    self.logger.info("Successfully built dimension tables")

    return self.dimension_tables

create_duckdb_dimension_tables 

create_duckdb_dimension_tables(table_prefix: Optional[str] = 'dim_', column_labels: Optional[Dict[str, str]] = None) -> None

Create dimension tables in DuckDB for categorical variables.

Parameters:

Name Type Description Default

table_prefix

Optional[str]

Prefix for dimension table names. Defaults to 'dim_'.

 'dim_'

column_labels

Optional[Dict[str, str]]

Optional mapping of column names to labels.

 None
Source code in dashboard_template_database/builders/tables.py 
 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
def create_duckdb_dimension_tables(self, table_prefix: Optional[str] = 'dim_', column_labels: Optional[Dict[str, str]] = None) -> None:
    """
    Create dimension tables in DuckDB for categorical variables.

    Args:
        table_prefix (Optional[str]): Prefix for dimension table names. Defaults to 'dim_'.
        column_labels (Optional[Dict[str, str]]): Optional mapping of column names to labels.

    """
    # Création du dictionnaire des tables de dimensions si elles n'existent pas déjà
    if not hasattr(self, 'dimension_tables'):
        _ = self.create_dimension_tables(column_labels)

    # Création de chaque table de dimension dans DuckDB
    for dim_name, dim_df in self.dimension_tables.items():
        # Initialisation du nom de la table
        table_name = f"{table_prefix}{dim_name}"
        # Enregistrement d'une vue temporaire
        self.conn.register('temp_dim', dim_df)

        # Création d'une table avec "value" comme clé primaire
        self.conn.execute(f"""
            CREATE TABLE {table_name} AS 
            SELECT
                value,
                label 
            FROM temp_dim
        """)

        # Ajout de la vue correspondante
        self.conn.execute('DROP VIEW temp_dim')

        # Logging
        self.logger.info(f"Successfully registered duckdb dimension table for {dim_name}")

create_duckdb_fact_table 

create_duckdb_fact_table(table_name: Optional[str] = 'fact_table', table_prefix: Optional[str] = 'dim_', column_labels: Optional[Dict[str, str]] = None) -> None

Create a fact table in DuckDB with foreign key relationships to dimension tables.

Parameters:

Name Type Description Default

table_name

Optional[str]

Name of the fact table in DuckDB. Defaults to 'fact_table'.

 'fact_table'

table_prefix

Optional[str]

Prefix for dimension table names. Defaults to 'dim_'.

 'dim_'

column_labels

Optional[Dict[str, str]]

Optional mapping of column names to labels.

 None
Source code in dashboard_template_database/builders/tables.py 
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
158
159
160
161
162
163
164
165
166
167
168
169
def create_duckdb_fact_table(self, table_name: Optional[str] = 'fact_table', table_prefix: Optional[str] = 'dim_', column_labels: Optional[Dict[str, str]] = None) -> None:
    """
    Create a fact table in DuckDB with foreign key relationships to dimension tables.

    Args:
        table_name (Optional[str]): Name of the fact table in DuckDB. Defaults to 'fact_table'.
        table_prefix (Optional[str]): Prefix for dimension table names. Defaults to 'dim_'.
        column_labels (Optional[Dict[str, str]]): Optional mapping of column names to labels.

    """
    # Création de la table d'informations si elle n'existe pas déjà
    if not hasattr(self, 'df_fact'):
        _ = self.create_fact_table(column_labels)

    # Enregistrement de la table comme vue temporaire
    self.conn.register('temp_fact', self.df_fact)

    # Initialisation de la liste des conditions de jointure avec les tables de dimensions
    join_conditions = []
    # Initialisation de la liste des colonnes à sélectionner
    # select_columns = []

    # Parcours des tables de dimensions correspondant à des clés étrangères dans la table d'informations
    for dim_name in self.dimension_tables.keys():
        # Initialisation du nom de la table
        dim_table = f"{table_prefix}{dim_name}"

        # Ajout des conditions de jointure sur la base de la colonne "value" des tables de dimensions
        join_conditions.append(
            f"LEFT JOIN {dim_table} ON temp_fact.{dim_name} = {dim_table}.value"
        )

        # Logging
        self.logger.info(f"Successfully created foreign key for dimension '{dim_name}'")
        # Ajout de la colonne qui correspond à une clé étrangère
        # select_columns.append(f"temp_fact.{dim_name}")

    # Ajout des colonnes des variables non catégorielles
    # select_columns.extend([f"temp_fact.{col}" for col in self.df_fact.columns if col not in self.dimension_tables.keys()])

    # Création de la table d'information
    query = f"""
    CREATE TABLE {table_name} AS 
    SELECT *
    FROM temp_fact
    {' '.join(join_conditions)}
    """

    self.conn.execute(query)

    # Ajout de la vue
    self.conn.execute('DROP VIEW temp_fact')

    # Logging
    self.logger.info(f"Successfully registered duckdb fact table")

create_duckdb_metadata_table 

create_duckdb_metadata_table(table_name: Optional[str] = 'metadata', column_labels: Optional[Dict[str, str]] = None) -> None

Create a metadata table in DuckDB.

Parameters:

Name Type Description Default

table_name

Optional[str]

Name of the metadata table in DuckDB. Defaults to 'metadata'.

 'metadata'

column_labels

Optional[Dict[str, str]]

Optional mapping of column names to labels.

 None
Source code in dashboard_template_database/builders/tables.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def create_duckdb_metadata_table(self, table_name: Optional[str] = 'metadata', column_labels: Optional[Dict[str, str]] = None) -> None:
    """
    Create a metadata table in DuckDB.

    Args:
        table_name (Optional[str]): Name of the metadata table in DuckDB. Defaults to 'metadata'.
        column_labels (Optional[Dict[str, str]]): Optional mapping of column names to labels.

    """
    # Création de la table des méta-données si elle n'existe pas déjà
    if not hasattr(self, 'df_metadata'):
        _ = self.create_metadata_table(column_labels)

    # Conversion DataFrame en table DuckDB
    self.conn.register('temp_metadata', self.df_metadata)
    self.conn.execute(f"""
        CREATE TABLE {table_name} AS 
        SELECT * FROM temp_metadata
    """)
    self.conn.execute('DROP VIEW temp_metadata')

    # Logging
    self.logger.info("Successfully registered duckdb meta-data table")

create_fact_table 

create_fact_table(column_labels: Optional[Union[Dict[str, str], None]] = None) -> DataFrame

Generate a fact table by replacing categorical values with corresponding IDs.

Parameters:

Name Type Description Default

column_labels

dict

A dictionary mapping column names to labels. Defaults to None.

 None

Returns:

Type Description
DataFrame

pd.DataFrame: The fact table with categorical values replaced by IDs.
Source code in dashboard_template_database/builders/schema.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def create_fact_table(self, column_labels : Optional[Union[Dict[str, str], None]]= None) -> pd.DataFrame:
    """
    Generate a fact table by replacing categorical values with corresponding IDs.

    Args:
        column_labels (dict, optional): A dictionary mapping column names to labels.
                                        Defaults to None.

    Returns:
        pd.DataFrame: The fact table with categorical values replaced by IDs.
    """
    # Création des tables de dimensions si ces-dernières n'existent pas
    if not hasattr(self, 'dimension_tables') :
        _ = self.create_dimension_tables(column_labels=column_labels)

    # Initialisation de la table des informations
    self.df_fact = self.df.copy()

    # Remplacement des labels par leur valeur
    for column in self.dimension_tables.keys() :
        # Construction du dictionnaire de passage
        dict_label_value = {d['label'] : d['value'] for d in self.dimension_tables[column].to_dict(orient='records')}
        # Remplacement des valeurs
        self.df_fact[column] = self.df_fact[column].replace(dict_label_value)
        # Logging
        self.logger.info(f"Successfully replace modalities by ids in column '{column}'")

    # Logging
    self.logger.info("Successfully built fact table")

    return self.df_fact

create_metadata_table 

create_metadata_table(column_labels: Optional[Union[Dict[str, str], None]] = None) -> Dict

Automatically infer metadata for the DataFrame's columns, including types, labels, and categorical attributes.

Parameters:

Name Type Description Default

column_labels

dict

A dictionary mapping column names to labels. Defaults to None.

 None

Returns:

Type Description
Dict

pd.DataFrame: A DataFrame containing metadata for each column in the input dataset.
Source code in dashboard_template_database/builders/schema.py 
 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
def create_metadata_table(self, column_labels : Optional[Union[Dict[str, str], None]]= None) -> Dict:
    """
    Automatically infer metadata for the DataFrame's columns, including types, labels, 
    and categorical attributes.

    Args:
        column_labels (dict, optional): A dictionary mapping column names to labels.
                                        Defaults to None.

    Returns:
        pd.DataFrame: A DataFrame containing metadata for each column in the input dataset.
    """
    # Initialisation de la liste des méta-données
    list_metadata = []
    # Parcours des colonnes du jeu de données
    for col in self.df.columns:
        # Extraction du type de la colonne
        dtype = str(self.df[col].dtype)
        # Initialisation des méta-données associées à la colonne
        if column_labels is not None :
            metadata = {
                'name': col,
                'label': column_labels[col] if col in column_labels.keys() else col.replace('_', ' ').title(),
                'python_type': dtype,
                'sql_type': self._map_python_to_sql_type(dtype),
                'is_categorical': False,
                # 'modalities': None
            }
        else :
            metadata = {
                'name': col,
                'label': col.replace('_', ' ').title(),
                'python_type': dtype,
                'sql_type': self._map_python_to_sql_type(dtype),
                'is_categorical': False,
                # 'modalities': None
            }

        # Logging
        self.logger.info(f"Successfully extracted meta-data from column '{col}'")

        # Si la colonne est object
        if dtype == 'object' :
            # Calcul du nombre de modalités
            n_modalities = self.df[col].nunique()
            # Si le nombre de modalités dans la colonne est inférieur au seuil, la variable est catégorielle
            if  n_modalities <= self.categorical_threshold:
                # Mise à jour du type de la variable
                metadata['is_categorical'] = True
                # Logging
                self.logger.info(f"The column '{col}' is of type 'object' and the number of modalities {n_modalities} satisfies the categorical threshold criteria {self.categorical_threshold}")
                # Mise à jour des modalités
                # metadata['modalities'] = str(self.df[col].dropna().unique().tolist())
            else :
                # Logging
                self.logger.warning(f"The column '{col}' is  of type 'object' but the number of modalities {n_modalities} exceeds the categorical threshold criteria {self.categorical_threshold}")

        # Ajout au dictionnaire
        list_metadata.append(metadata)            

    # Création d'un DataFrame
    self.df_metadata = pd.DataFrame.from_dict(list_metadata).sort_values(by='label', ascending=True, ignore_index=True)

    # Logging
    self.logger.info("Successfully built the meta-data DataFrame")

    return self.df_metadata

display_schema 

display_schema() -> None

Display the structure of all tables in the DuckDB schema.

Source code in dashboard_template_database/builders/tables.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def display_schema(self) -> None:
    """
    Display the structure of all tables in the DuckDB schema.
    """
    # Extraction des tables
    tables = self.conn.execute("SHOW TABLES").fetchall()
    # Logging
    self.logger.info("\n Created Tables:")
    # Parcours des tables
    for table in tables:
        # Affichage de la structure
        self.logger.info(f"\n {table[0]} Structure:")
        # Extraction des informations relatives à la table
        table_info = self.conn.execute(f"DESCRIBE {table[0]}").fetchall()
        # Affichage de chaque information
        for col in table_info:
            self.logger.info(f"  {col[0]}: {col[1]}")