API reference

Everything most code needs is importable from pyrosome itself, e.g. from pyrosome import ParquetDal.

Data access

class pyrosome.Dal(*args, **kwargs)[source]

Bases: Protocol

Versioned, append-only storage for data frames, whatever the file format.

Example

>>> from pathlib import Path
>>> import polars as pl
>>> from pyrosome import CsvDal, Dal, ParquetDal
>>> def save(dal: Dal) -> str:
...     return dal.write(pl.DataFrame({"x": [1]}), "example").name
>>> save(ParquetDal(Path("parquet"))), save(CsvDal(Path("csv")))
('v0', 'v0')
read(*keys)[source]

Read data from the latest version of the data described by keys.

Parameters:

*keys (str) – Path components naming the profile.

Returns:

A lazy scan over every data file in the latest version.

Raises:

FileNotFoundError – If there is no data for keys.

Return type:

LazyFrame

write(data, *keys)[source]

Write the data as a new version.

Parameters:

  • data (DataFrame | LazyFrame) – The data to write.

  • *keys (str) – Path components naming the profile.

Returns:

The versioned directory the data was written to.

Return type:

Path

append(data, *keys)[source]

Add a new file to the latest version without touching existing files.

Parameters:

  • data (DataFrame | LazyFrame) – The data to append. Its columns must match the existing data.

  • *keys (str) – Path components naming the profile.

Returns:

The path of the new file, or the new versioned directory if there was no existing data.

Return type:

Path

class pyrosome.ParquetDal(data_directory)[source]

Bases: _FileDal

Data Access Layer storing versioned data as parquet files.

Each profile, named by one or more keys, is a directory under data_directory. Every write creates a new version and append adds files to the latest version, so existing data is never modified in place.

Example

>>> from pathlib import Path
>>> import polars as pl
>>> from pyrosome import ParquetDal
>>> dal = ParquetDal(Path("data"))
>>> dal.write(pl.DataFrame({"x": [1, 2]}), "example").name
'v0'
>>> dal.append(pl.DataFrame({"x": [3]}), "example").name
'1.parquet'
>>> dal.read("example").collect()["x"].to_list()
[1, 2, 3]

Create a data access layer. Nothing is read or written until used.

Parameters:

data_directory (Path) – The directory where data files are stored.

Returns:

None.

extension: str = 'parquet'

File extension of the data files, without the leading dot.

append(data, *keys)

Add a new file to the latest data directory without touching existing files.

Columns are reordered and cast to match the existing data. If there is no existing data this warns and writes instead.

Parameters:

  • data (DataFrame | LazyFrame) – The data to append. Its columns must match the existing data.

  • *keys (str) – Path components naming the profile.

Returns:

The path of the new file, or the new versioned directory if there was no existing data.

Raises:
Return type:

Path

get_profile_dir(*keys)

Get path to the profile directory described by keys.

Parameters:

*keys (str) – Path components naming the profile.

Returns:

The profile directory.

Raises:

ValueError – If there are no keys or they lead outside the data directory.

Return type:

Path

read(*keys)

Read data from the latest version of the data described by keys.

Parameters:

*keys (str) – Path components naming the profile.

Returns:

A lazy scan over every data file in the latest version.

Raises:
Return type:

LazyFrame

write(data, *keys)

Write the data as a new version.

Readers keep seeing the previous version until the write is complete.

Parameters:

  • data (DataFrame | LazyFrame) – The data to write.

  • *keys (str) – Path components naming the profile.

Returns:

The versioned directory the data was written to.

Raises:
Return type:

Path

data_directory: Path

The directory where data files are stored.

class pyrosome.CsvDal(data_directory)[source]

Bases: _FileDal

Same as a ParquetDal but uses csv files instead of parquet files.

Broadly recommend using the parquet version, but this exists for user convenience. Column types are inferred when the files are read back.

Example

>>> from pathlib import Path
>>> import polars as pl
>>> from pyrosome import CsvDal
>>> csv_dal = CsvDal(Path("data"))
>>> output_dir = csv_dal.write(pl.DataFrame({"x": [1, 2]}), "example_csv")
>>> (output_dir / "0.csv").read_text().splitlines()
['x', '1', '2']

Create a data access layer. Nothing is read or written until used.

Parameters:

data_directory (Path) – The directory where data files are stored.

Returns:

None.

extension: str = 'csv'

File extension of the data files, without the leading dot.

append(data, *keys)

Add a new file to the latest data directory without touching existing files.

Columns are reordered and cast to match the existing data. If there is no existing data this warns and writes instead.

Parameters:

  • data (DataFrame | LazyFrame) – The data to append. Its columns must match the existing data.

  • *keys (str) – Path components naming the profile.

Returns:

The path of the new file, or the new versioned directory if there was no existing data.

Raises:
Return type:

Path

get_profile_dir(*keys)

Get path to the profile directory described by keys.

Parameters:

*keys (str) – Path components naming the profile.

Returns:

The profile directory.

Raises:

ValueError – If there are no keys or they lead outside the data directory.

Return type:

Path

read(*keys)

Read data from the latest version of the data described by keys.

Parameters:

*keys (str) – Path components naming the profile.

Returns:

A lazy scan over every data file in the latest version.

Raises:
Return type:

LazyFrame

write(data, *keys)

Write the data as a new version.

Readers keep seeing the previous version until the write is complete.

Parameters:

  • data (DataFrame | LazyFrame) – The data to write.

  • *keys (str) – Path components naming the profile.

Returns:

The versioned directory the data was written to.

Raises:
Return type:

Path

data_directory: Path

The directory where data files are stored.

pyrosome.get_accessor(file_ending: Literal['parquet']) → type[ParquetDal][source]
pyrosome.get_accessor(file_ending: Literal['csv']) → type[CsvDal]

Get the accessor class that stores files with the given file ending.

Parameters:

file_ending (str) – The file format to store, without a leading dot.

Returns:

ParquetDal for "parquet" or CsvDal for "csv".

Raises:

ValueError – If file_ending is not a supported format.

Return type:

type[ParquetDal | CsvDal]

Filtering reads

class pyrosome.SimpleIo(data_directory)[source]

Bases: object

Parquet storage whose read can also filter rows by column values.

Example

>>> from pathlib import Path
>>> import polars as pl
>>> from pyrosome import SimpleIo
>>> io = SimpleIo(Path("data"))
>>> df = pl.DataFrame({"x": [1, 2, 2], "y": ["a", "b", None]})
>>> io.write(df, "example").name
'v0'
>>> io.read("example", x=2, y="b").collect().rows()
[(2, 'b')]
>>> io.read("example", y=None).collect().rows()
[(2, None)]

write and append also take pandas data frames:

>>> import pandas as pd
>>> io.append(pd.DataFrame({"x": [3], "y": ["d"]}), "example").name
'1.parquet'
>>> io.read("example", x=3).collect().rows()
[(3, 'd')]

Create the inner dal. Nothing is read or written until used.

Parameters:

data_directory (Path | __annotationlib_name_1__ | os.PathLike[__annotationlib_name_2__]) – The directory where data files are stored.

Returns:

None.

dal: ParquetDal

The data access layer reads and writes go through.

available_profiles(*keys)[source]

List the profiles directly below keys that have been written.

A profile is a directory holding a manifest. Nested profiles, and directories that only group other profiles, are not listed; pass their path as keys to list what they hold.

Example

>>> from pathlib import Path
>>> import polars as pl
>>> from pyrosome import SimpleIo
>>> io = SimpleIo(Path("data"))
>>> df = pl.DataFrame({"x": [1]})
>>> _ = io.write(df, "shop", "orders")
>>> _ = io.write(df, "shop", "totals")
>>> io.available_profiles("shop")
('orders', 'totals')
>>> io.available_profiles("missing")
()
Parameters:

*keys (str) – Path components naming the directory to look in. With no keys, the data directory itself.

Returns:

The names of the profiles below keys, sorted. Empty if the directory doesn’t exist.

Raises:

  • ValueError – If keys lead outside the data directory.

  • OSError – If the directory cannot be listed.

Return type:

tuple[str, …]

read(*keys, **filters)[source]

Read a lazy frame, keeping only rows where each column equals a value.

Filters are equality only. Pass a column whose name isn’t a valid keyword argument by unpacking a dict, e.g. io.read("sales", **{"unit price": 3}).

Parameters:

  • *keys (str) – Path components naming the profile.

  • **filters (FilterValue) – Column names and the value each column must equal. None matches nulls.

Returns:

A lazy scan over the latest version, keeping only rows that match every filter.

Raises:
Return type:

LazyFrame

write(data, *keys)[source]

Write the data as a new version.

A pandas data frame’s named index levels are stored as columns; an unnamed index is dropped.

Parameters:

  • data (DataFrame | LazyFrame | pd.DataFrame) – The data to write, as a polars or pandas frame.

  • *keys (str) – Path components naming the profile.

Returns:

The versioned directory the data was written to.

Raises:
Return type:

Path

append(data, *keys)[source]

Add a new file to the latest version without touching existing files.

A pandas data frame’s named index levels are stored as columns; an unnamed index is dropped.

Parameters:

  • data (DataFrame | LazyFrame | pd.DataFrame) – The data to append, as a polars or pandas frame. Its columns must match the existing data.

  • *keys (str) – Path components naming the profile.

Returns:

The path of the new file, or the new versioned directory if there was no existing data.

Raises:
Return type:

Path

pyrosome.api.FilterValue = FilterValue

A value pyrosome.SimpleIo.read can filter a column by.

datetime.datetime is a datetime.date, so both are accepted. Floats are not: equality on floats is unreliable, since 0.1 + 0.2 != 0.3.

Datasets

class pyrosome.Dataset(dal, *keys, **filters)[source]

Bases: object

Represents a dataset.

A dataset is defined by a set of keys, which point to specific folders in the data directory. This usually will be a system that contains profiles, but a user could add more nesting if they so desired.

Every profile read through a dataset is filtered by the dataset’s filters, so every profile is expected to have the filtered columns.

Subclasses can derive profiles from other profiles: a method named _calc_<profile> (keys joined by underscores) returns the profile’s data, and get_cache_profile writes it when the profile has no data yet. The method is called on the dataset without_filters, so it computes the profile for every row and the cache is shared by all subsets.

Example

>>> from pathlib import Path
>>> import polars as pl
>>> from pyrosome import Dataset, SimpleIo
>>> class Shop(Dataset):
...     def _calc_totals(self) -> pl.LazyFrame:
...         orders = self.get_profile("orders")
...         return orders.group_by("region").agg(pl.col("amount").sum())
>>> io = SimpleIo(Path("data"))
>>> orders = pl.DataFrame({"region": ["eu", "us", "eu"], "amount": [1, 2, 3]})
>>> io.write(orders, "shop", "orders").name
'v0'
>>> eu = Shop(io, "shop", region="eu")
>>> eu.get_profile("orders").collect()["amount"].to_list()
[1, 3]
>>> eu.get_cache_profile("totals").collect().rows()
[('eu', 4)]
>>> io.read("shop", "totals").collect().height
2

Set up the dataset. Nothing is read until a profile is requested.

Subclasses should keep this signature: get_subset and without_filters call it to make datasets of the same class.

Parameters:

  • dal (SimpleIo) – The storage the dataset’s profiles are read from.

  • *keys (str) – Path components naming the dataset’s directory.

  • **filters (FilterValue) – Column names and the value each column must equal in every profile read through this dataset. None matches nulls.

Returns:

None.

keys: tuple[str, ...]

Keys which define this dataset.

filters: Mapping[str, FilterValue]

Row-wise filters to apply when loading the data.

get_profile(*keys)[source]

Get a profile defined by keys, filtered by the dataset’s filters.

Parameters:

*keys (str) – Path components naming the profile, below the dataset’s keys.

Returns:

A lazy scan over the profile’s latest version, keeping only rows that match every filter.

Raises:
Return type:

pl.LazyFrame

get_cache_profile(*keys, refresh=False)[source]

Get a profile, or calculate and cache it if not found.

The profile is calculated by the _calc_<profile> method on the dataset without_filters, and written as a new version, so the cached profile holds every row. The dataset’s filters are applied when it is read.

Parameters:

  • *keys (str) – Path components naming the profile, below the dataset’s keys.

  • refresh (bool) – Recalculate the profile even if it has data, e.g. after the profiles it is calculated from have changed.

Returns:

A lazy scan over the profile’s latest version, keeping only rows that match every filter.

Raises:
Return type:

pl.LazyFrame

without_filters()[source]

Get this dataset without any filters.

Returns:

A dataset of the same class, with the same keys and no filters.

Return type:

Self

get_subset(**new_filters)[source]

Apply additional filters to further reduce this dataset.

Parameters:

**new_filters (FilterValue) – Column names and the value each column must equal, on top of this dataset’s filters.

Returns:

A dataset of the same class, with the same keys and the combined filters.

Raises:

ValueError – If this dataset already filters one of the columns.

Return type:

Self

Internals

These live in submodules and are rarely needed directly.

Primary API for writing/reading/appending.

class pyrosome.dal.DataManifestFile(profile_path)[source]

Bases: object

A manifest file that records the latest version of a profile.

Create a handle on the manifest in profile_path. Nothing is read yet.

Parameters:

profile_path (Path) – Path to the profile directory.

Returns:

None.

MANIFEST_FILE_NAME: Final[str] = 'manifest.json'

Name of the manifest file inside the profile directory.

LOCK_TIMEOUT: Final[float] = 10.0

Seconds to wait for another process to finish writing the manifest.

profile_path: Path

Path to the profile directory.

exists()[source]

Does the manifest file even exist yet?

Returns:

Whether the manifest file exists.

Return type:

bool

get_latest_data_dir()[source]

Get the latest data directory described by the manifest file.

Returns:

The latest data directory.

Raises:

  • FileNotFoundError – If the manifest file or the latest data directory does not exist.

  • TypeError – If the manifest file is corrupt.

Return type:

Path

reserve_new_data_dir()[source]

Create and return the next version of the data directory.

The manifest is not updated; call commit_data_dir once the data is written.

Returns:

The new, empty data directory.

Raises:

OSError – If the profile directory cannot be listed or written to.

Return type:

Path

commit_data_dir(data_dir)[source]

Point the manifest at data_dir, unless it already points at a newer version.

Parameters:

data_dir (Path) – The finished data directory.

Returns:

None.

Raises:
Return type:

None

Create a simple lock on a file.

Nothing can read or write to a file while it is locked. This prevents multiple processes from reading or writing to the same file at the same time, which could cause data corruption.

exception pyrosome.simple_file_lock.LockError[source]

Bases: Exception

An error that occurs during file locking.

exception pyrosome.simple_file_lock.FileAlreadyLockedError[source]

Bases: LockError

An error that occurs when a file is already locked and cannot be locked again.

exception pyrosome.simple_file_lock.FileLockPermissionError[source]

Bases: LockError

An error that occurs when a file cannot be locked due to permission issues.

exception pyrosome.simple_file_lock.FileLockedError[source]

Bases: LockError

An error that occurs when a file is locked and cannot be accessed.

class pyrosome.simple_file_lock.SimpleWriteLock(path, timeout=0.0, poll_interval=0.05)[source]

Bases: object

Simple lock that prevents other processes accessing a file being written to.

The lock is a <file>.write_lock file next to the locked file, so it works across processes but is only advisory: code that ignores the lock is not blocked.

Example

>>> from pathlib import Path
>>> from pyrosome.simple_file_lock import SimpleWriteLock, file_is_locked
>>> target = Path("notes.txt")
>>> with SimpleWriteLock(target, timeout=1.0):
...     file_is_locked(target)
True
>>> file_is_locked(target)
False

Create a lock for path. The lock is only taken when entering the context.

Parameters:

  • path (Path) – The file to lock.

  • timeout (float) – Seconds to keep retrying if another process holds the lock.

  • poll_interval (float) – Seconds to wait between attempts to take the lock.

Returns:

None.

Raises:

ValueError – If path has an empty name, such as /.

timeout

Seconds to keep retrying if another process holds the lock.

pyrosome.simple_file_lock.file_is_locked(path)[source]

Check if a file is locked.

Parameters:

path (Path) – The file to check.

Returns:

Whether a lock is currently held on path.

Raises:

ValueError – If path has an empty name, such as /.

Return type:

bool