Created
October 10, 2022 14:26
-
-
Save clbarnes/9c53701e8b436603d236ac0ab26ff8a9 to your computer and use it in GitHub Desktop.
Build dataframes row-wise in an ergonomic and reasonably efficient way
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """Module for building pandas DataFrames by row.""" | |
| from collections.abc import Sequence, Collection | |
| import typing as tp | |
| import pandas as pd | |
| class DataFrameBuilder: | |
| def __init__( | |
| self, columns: Sequence[tp.Hashable], dtypes: tp.Optional[Sequence] = None | |
| ): | |
| self.columns: dict[tp.Hashable, list] = {c: [] for c in columns} | |
| self.dtypes: tp.Optional[list] = list(dtypes) if dtypes else None | |
| if dtypes is not None and len(dtypes) != len(self.columns): | |
| raise ValueError() | |
| def _check_len(self, row: Collection): | |
| """Raise an error if row is of incorrect length. | |
| Parameters | |
| ---------- | |
| row : Collection | |
| Row to be added (as dict or sequence). | |
| Raises | |
| ------ | |
| ValueError | |
| If row length does not match number of columns. | |
| """ | |
| if len(row) != len(self.columns): | |
| raise ValueError( | |
| f"Row length ({len(row)}) does not match number of columns ({len(self.columns)})" | |
| ) | |
| def append_row(self, row: Sequence): | |
| """Append a sequence to the rows. | |
| Parameters | |
| ---------- | |
| row : Sequence | |
| Must be same length as number of columns. | |
| Returns | |
| ------- | |
| self | |
| """ | |
| self._check_len(row) | |
| for item, col in zip(row, self.columns.values()): | |
| col.append(item) | |
| return self | |
| def append_dict(self, row: dict[tp.Hashable, tp.Any]): | |
| """Append a dict to the rows. | |
| Parameters | |
| ---------- | |
| row : dict[tp.Hashable, tp.Any] | |
| Keys must match columns. | |
| Returns | |
| ------- | |
| self | |
| """ | |
| self._check_len(row) | |
| for k, v in row.items(): | |
| self.columns[k].append(v) | |
| return self | |
| def build(self, index_col=None) -> pd.DataFrame: | |
| """Build the dataframe. | |
| Parameters | |
| ---------- | |
| index_col : Hashable, optional | |
| Which column to use as the index, by default None | |
| (i.e. numeric index in insertion order). | |
| Returns | |
| ------- | |
| pandas.DataFrame | |
| """ | |
| cols = dict() | |
| index = None | |
| for idx, (k, v) in enumerate(self.columns.items()): | |
| dtype = self.dtypes[idx] if self.dtypes else None | |
| v2 = pd.Series(v, dtype=dtype, name=k) | |
| if k == index_col: | |
| index = v2 | |
| else: | |
| cols[k] = v2 | |
| df = pd.DataFrame.from_dict(cols) | |
| if index is not None: | |
| df.index = index | |
| return df |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment