"""Utils required for dataframes.""" from typing import Dict, List, Optional import pandas as pd def dict_to_dataframe(records: List[Dict[str, int | bool]], dtypes=None) -> pd.DataFrame: """Convert a list of dictionaries to a pandas DataFrame. Args: records (List[Dict[str, Any]]): List of dictionaries to convert. dtypes (Dict[str, str] | None): dictionary specifying column data types. Returns: pd.DataFrame: DataFrame created from the records. """ if not records: return pd.DataFrame() df = pd.DataFrame(records) if dtypes: df = df.astype(dtypes) return df def merge_dataframes( dataframes: List[pd.DataFrame], columns: Optional[List[str]] = None, drop_duplicates: bool = False, ignore_index: bool = True, ) -> pd.DataFrame: """Merge multiple pandas DataFrames into one, optionally selecting columns and dropping duplicates. Args: dataframes (List[pd.DataFrame]): List of DataFrames to concatenate. columns (Optional[List[str]]): Columns to select after concatenation. Defaults to None (all columns). drop_duplicates (bool): Whether to drop duplicate rows after concatenation. Defaults to False. ignore_index (bool): Whether to reset the index after concatenation. Defaults to True. Returns: pd.DataFrame: The merged DataFrame. """ non_empty_dfs = [dataframe for dataframe in dataframes if not dataframe.empty] if not non_empty_dfs: return pd.DataFrame(columns=columns if columns else []) merged_df = pd.concat(non_empty_dfs, ignore_index=ignore_index) if columns: merged_df = merged_df.loc[:, columns] if drop_duplicates: merged_df = merged_df.drop_duplicates() return merged_df