Skip to content

Instantly share code, notes, and snippets.

@clbarnes
Created October 10, 2022 14:26
Show Gist options
  • Select an option

  • Save clbarnes/9c53701e8b436603d236ac0ab26ff8a9 to your computer and use it in GitHub Desktop.

Select an option

Save clbarnes/9c53701e8b436603d236ac0ab26ff8a9 to your computer and use it in GitHub Desktop.
Build dataframes row-wise in an ergonomic and reasonably efficient way
#!/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