Store

class chronify.store.Store(backend: IbisBackend | None = None, backend_name: str | None = None, file_path: Path | str | None = None)

Data store for time series data

Construct the Store.

Parameters:

• backend – Optional, defaults to a DuckDB in-memory backend.

• backend_name – Optional, name of backend to use (‘duckdb’, ‘sqlite’). Mutually exclusive with backend.

• file_path – Optional, use this file for the database. If the file does not exist, create a new database. If the file exists, load that existing database. Defaults to a new in-memory database.

classmethod create_in_memory_db(backend_name: str = 'duckdb') → Store

Create a Store with an in-memory database.

classmethod create_file_db(file_path: Path | str = 'time_series.db', backend_name: str = 'duckdb', overwrite: bool = False) → Store

Create a Store with a file-based database.

classmethod load_from_file(file_path: Path | str, backend_name: str = 'duckdb') → Store

Load an existing store from a database.

dispose() → None

Dispose of the current connections.

get_table(name: str) → Table

Return the ibis Table expression.

has_table(name: str) → bool

Return True if the database has a table with the given name.

list_tables() → list[str]

Return a list of user tables in the database.

try_get_table(name: str) → Table | None

Return the ibis Table expression or None if it is not stored.

backup(dst: Path | str, overwrite: bool = False) → None

Copy the database to a new location. Not yet supported for in-memory databases.

property backend: IbisBackend

Return the ibis backend.

property schema_manager: SchemaManager

Return the store’s schema manager.

check_timestamps(name: str) → None

Check the timestamps in the table.

Parameters:

name – Name of the table to check.

Raises:

InvalidTable – Raised if the timestamps do not match the schema.

create_view_from_parquet(path: Path, schema: TableSchema, bypass_checks: bool = False) → None

Load a table into the database from a Parquet file.

ingest_from_csv(path: Path | str, src_schema: CsvTableSchema, dst_schema: TableSchema) → bool

Ingest data from a CSV file.

Parameters:

• path – Source data file

• src_schema – Defines the schema of the source file.

• dst_schema – Defines the destination table in the database.

Returns:

Return True if a table was created.

Return type:

bool

Raises:

InvalidTable – Raised if the data does not match the schema.

Examples

>>> resolution = timedelta(hours=1)
>>> time_config = DatetimeRange(
...     time_column="timestamp",
...     start=datetime(2020, 1, 1, 0),
...     length=8784,
...     resolution=timedelta(hours=1),
... )
>>> store = Store()
>>> store.ingest_from_csv(
...     "data.csv",
...     CsvTableSchema(
...         time_config=time_config,
...         pivoted_dimension_name="device",
...         value_columns=["device1", "device2", "device3"],
...     ),
...     TableSchema(
...         name="devices",
...         value_column="value",
...         time_config=time_config,
...         time_array_id_columns=["device"],
...     ),
... )

See also

ingest_from_csvs

ingest_from_csvs(paths: Iterable[Path | str], src_schema: CsvTableSchema, dst_schema: TableSchema) → bool

Ingest data into the table specified by schema. If the table does not exist, create it. This is faster than calling ingest_from_csv() many times. Each file is loaded into memory one at a time.

On DuckDB and SQLite, a failure rolls back every change made by this call and the database state is restored. On Spark, partial appends to a pre-existing table cannot be rolled back; only the case where this call creates a new table is fully cleaned up on failure.

Parameters:

• paths – Source data files

• src_schema – Defines the schema of the source files.

• dst_schema – Defines the destination table in the database.

Returns:

Return True if a table was created.

Return type:

bool

Raises:

InvalidTable – Raised if the data does not match the schema.

See also

ingest_from_csv

ingest_pivoted_table(data: DataFrame | Table, src_schema: PivotedTableSchema | CsvTableSchema, dst_schema: TableSchema) → bool

Ingest pivoted data into the table specified by schema. If the table does not exist, create it. Chronify will unpivot the data before ingesting it.

Parameters:

• data – Input data to ingest into the database.

• src_schema – Defines the schema of the input data.

• dst_schema – Defines the destination table in the database.

Returns:

Return True if a table was created.

Return type:

bool

Raises:

InvalidTable – Raised if the data does not match the schema.

Examples

>>> resolution = timedelta(hours=1)
>>> df = pd.DataFrame(
...     {
...         "timestamp": pd.date_range(
...             "2020-01-01", "2020-12-31 23:00:00", freq=resolution
...         ),
...         "device1": np.random.random(8784),
...         "device2": np.random.random(8784),
...         "device3": np.random.random(8784),
...     }
... )
>>> time_config = DatetimeRange(
...     time_column="timestamp",
...     start=datetime(2020, 1, 1, 0),
...     length=8784,
...     resolution=timedelta(hours=1),
... )
>>> store = Store()
>>> store.ingest_pivoted_table(
...     df,
...     PivotedTableSchema(
...         time_config=time_config,
...         pivoted_dimension_name="device",
...         value_columns=["device1", "device2", "device3"],
...     ),
...     TableSchema(
...         name="devices",
...         value_column="value",
...         time_config=time_config,
...         time_array_id_columns=["device"],
...     ),
... )

See also

ingest_pivoted_tables

ingest_pivoted_tables(data: Iterable[DataFrame | Table], src_schema: PivotedTableSchema | CsvTableSchema, dst_schema: TableSchema) → bool

Ingest pivoted data into the table specified by schema.

If the table does not exist, create it. Unpivot the data before ingesting it. This is faster than calling ingest_pivoted_table() many times.

On DuckDB and SQLite, a failure rolls back every change made by this call and the database state is restored. On Spark, partial appends to a pre-existing table cannot be rolled back; only the case where this call creates a new table is fully cleaned up on failure.

Parameters:

• data – Data to ingest into the database.

• src_schema – Defines the schema of all input tables.

• dst_schema – Defines the destination table in the database.

Returns:

Return True if a table was created.

Return type:

bool

See also

ingest_pivoted_table

ingest_table(data: DataFrame | Table, schema: TableSchema, **kwargs: Any) → bool

Ingest data into the table specified by schema. If the table does not exist, create it.

Parameters:

• data – Input data to ingest into the database.

• schema – Defines the destination table in the database.

Returns:

Return True if a table was created.

Return type:

bool

Raises:

InvalidTable – Raised if the data does not match the schema.

Examples

>>> store = Store()
>>> resolution = timedelta(hours=1)
>>> df = pd.DataFrame(
...     {
...         "timestamp": pd.date_range(
...             "2020-01-01", "2020-12-31 23:00:00", freq=resolution
...         ),
...         "value": np.random.random(8784),
...     }
... )
>>> df["id"] = 1
>>> store.ingest_table(
...     df,
...     TableSchema(
...         name="devices",
...         value_column="value",
...         time_config=DatetimeRange(
...             time_column="timestamp",
...             start=datetime(2020, 1, 1, 0),
...             length=8784,
...             resolution=timedelta(hours=1),
...         ),
...         time_array_id_columns=["id"],
...     ),
... )

See also

ingest_tables

ingest_tables(data: Iterable[DataFrame | Table], schema: TableSchema, **kwargs: Any) → bool

Ingest multiple input tables to the same database table. All tables must have the same schema. This offers significant performance advantages over calling ingest_table() many times.

On DuckDB and SQLite, a failure rolls back every change made by this call and the database state is restored. On Spark, partial appends to a pre-existing table cannot be rolled back; only the case where this call creates a new table is fully cleaned up on failure.

Parameters:

• data – Input tables to ingest into one database table.

• schema – Defines the destination table.

Returns:

Return True if a table was created.

Return type:

bool

Raises:

InvalidTable – Raised if the data does not match the schema.

See also

ingest_table

map_table_time_config(src_name: str, dst_schema: TableSchema, data_adjustment: TimeBasedDataAdjustment | None = None, wrap_time_allowed: bool = False, output_file: Path | None = None, check_mapped_timestamps: bool = False) → None

Map the existing table represented by src_name to a new table represented by dst_schema with a different time configuration.

Parameters:

• src_name – Refers to the table name of the source data.

• dst_schema – Defines the table to create in the database. Must not already exist.

• data_adjustment – Defines how the dataframe may need to be adjusted with respect to time. Data is only adjusted when the conditions apply.

• wrap_time_allowed – Defines whether the time column is allowed to be wrapped according to the time config in dst_schema when it does not line up with the time config

• output_file – If set, write the mapped table to this Parquet file.

• check_mapped_timestamps – Perform time checks on the result of the mapping operation. This can be slow and is not required.

Raises:

• InvalidTable – Raised if the schemas are incompatible.

• TableAlreadyExists – Raised if the dst_schema name already exists.

Examples

>>> store = Store()
>>> hours_per_year = 12 * 7 * 24
>>> num_time_arrays = 3
>>> df = pd.DataFrame(
...     {
...         "id": np.concatenate(
...             [np.repeat(i, hours_per_year) for i in range(1, 1 + num_time_arrays)]
...         ),
...         "month": np.tile(np.repeat(range(1, 13), 7 * 24), num_time_arrays),
...         "day_of_week": np.tile(np.tile(np.repeat(range(7), 24), 12), num_time_arrays),
...         "hour": np.tile(np.tile(range(24), 12 * 7), num_time_arrays),
...         "value": np.random.random(hours_per_year * num_time_arrays),
...     }
... )
>>> schema = TableSchema(
...     name="devices_by_representative_time",
...     value_column="value",
...     time_config=RepresentativePeriodTimeNTZ(
...         time_format=RepresentativePeriodFormat.ONE_WEEK_PER_MONTH_BY_HOUR,
...     ),
...     time_array_id_columns=["id"],
... )
>>> store.ingest_table(df, schema)
>>> store.map_table_time_config(
...     "devices_by_representative_time",
...     TableSchema(
...         name="devices_by_datetime",
...         value_column="value",
...         time_config=DatetimeRange(
...             time_column="timestamp",
...             start=datetime(2020, 1, 1, 0),
...             length=8784,
...             resolution=timedelta(hours=1),
...         ),
...         time_array_id_columns=["id"],
...     ),
... )

convert_time_zone(src_name: str, time_zone: tzinfo | None, output_file: Path | None = None, check_mapped_timestamps: bool = False) → TableSchema

Convert the time zone of the existing table represented by src_name to a new time zone.

Parameters:

• src_name – Refers to the table name of the source data.

• time_zone – Time zone to convert to.

• output_file – If set, write the mapped table to this Parquet file.

• check_mapped_timestamps – Perform time checks on the result of the mapping operation. This can be slow and is not required.

Raises:

TableAlreadyExists – Raised if the dst_schema name already exists.

Examples

>>> store = Store()
>>> start = datetime(year=2018, month=1, day=1, tzinfo=ZoneInfo("Etc/GMT+5"))
>>> freq = timedelta(hours=1)
>>> hours_per_year = 8760
>>> num_time_arrays = 1
>>> df = pd.DataFrame(
...     {
...         "id": np.concatenate(
...             [np.repeat(i, hours_per_year) for i in range(1, 1 + num_time_arrays)]
...         ),
...         "timestamp": np.tile(
...             pd.date_range(start, periods=hours_per_year, freq="h"), num_time_arrays
...         ),
...         "value": np.random.random(hours_per_year * num_time_arrays),
...     }
... )
>>> schema = TableSchema(
...     name="some_data",
...     time_config=DatetimeRange(
...         time_column="timestamp",
...         start=start,
...         length=hours_per_year,
...         resolution=freq,
...     ),
...     time_array_id_columns=["id"],
...     value_column="value",
... )
>>> store.ingest_table(df, schema)
>>> to_time_zone = ZoneInfo("US/Mountain")
>>> dst_schema = store.convert_time_zone(
...     schema.name, to_time_zone, check_mapped_timestamps=True
... )

convert_time_zone_by_column(src_name: str, time_zone_column: str, wrap_time_allowed: bool = False, output_file: Path | None = None, check_mapped_timestamps: bool = False) → TableSchema

Convert the time zone of the existing table represented by src_name to new time zone(s) defined by a column

Parameters:

• src_name – Refers to the table name of the source data.

• time_zone_column – Name of the time zone column for conversion.

• wrap_time_allowed – Defines whether the time column is allowed to be wrapped to reflect the same time range as the src_name schema in tz-naive clock time

• output_file – If set, write the mapped table to this Parquet file.

• check_mapped_timestamps – Perform time checks on the result of the mapping operation. This can be slow and is not required.

Raises:

TableAlreadyExists – Raised if the dst_schema name already exists.

Examples

>>> store = Store()
>>> start = datetime(year=2018, month=1, day=1, tzinfo=ZoneInfo("Etc/GMT+5"))
>>> freq = timedelta(hours=1)
>>> hours_per_year = 8760
>>> num_time_arrays = 3
>>> df = pd.DataFrame(
...     {
...         "id": np.concatenate(
...             [np.repeat(i, hours_per_year) for i in range(1, 1 + num_time_arrays)]
...         ),
...         "timestamp": np.tile(
...             pd.date_range(start, periods=hours_per_year, freq="h"), num_time_arrays
...         ),
...         "time_zone": np.repeat(["US/Eastern", "US/Mountain", "None"], hours_per_year),
...         "value": np.random.random(hours_per_year * num_time_arrays),
...     }
... )
>>> schema = TableSchema(
...     name="some_data",
...     time_config=DatetimeRange(
...         time_column="timestamp",
...         start=start,
...         length=hours_per_year,
...         resolution=freq,
...     ),
...     time_array_id_columns=["id"],
...     value_column="value",
... )
>>> store.ingest_table(df, schema)
>>> time_zone_column = "time_zone"
>>> dst_schema = store.convert_time_zone_by_column(
...     schema.name,
...     time_zone_column,
...     wrap_time_allowed=False,
...     check_mapped_timestamps=True,
... )

localize_time_zone(src_name: str, time_zone: tzinfo | None, output_file: Path | None = None, check_mapped_timestamps: bool = False) → TableSchema

Localize the time zone of the existing table represented by src_name to a specified time zone

Parameters:

• src_name – Refers to the table name of the source data.

• time_zone – Standard time zone to localize to. If None, keep as tz-naive.

• output_file – If set, write the mapped table to this Parquet file.

• check_mapped_timestamps – Perform time checks on the result of the mapping operation. This can be slow and is not required.

Raises:

TableAlreadyExists – Raised if the dst_schema name already exists.

Returns:

The schema of the newly created table.

Return type:

TableSchema

Examples

>>> store = Store()
>>> start = datetime(year=2018, month=1, day=1)  # tz-naive
>>> freq = timedelta(hours=1)
>>> hours_per_year = 8760
>>> num_time_arrays = 1
>>> df = pd.DataFrame(
...     {
...         "id": np.concatenate(
...             [np.repeat(i, hours_per_year) for i in range(1, 1 + num_time_arrays)]
...         ),
...         "timestamp": np.tile(
...             pd.date_range(start, periods=hours_per_year, freq="h"), num_time_arrays
...         ),
...         "value": np.random.random(hours_per_year * num_time_arrays),
...     }
... )
>>> schema = TableSchema(
...     name="some_data",
...     time_config=DatetimeRange(
...         time_column="timestamp",
...         start=start,
...         length=hours_per_year,
...         resolution=freq,
...     ),
...     time_array_id_columns=["id"],
...     value_column="value",
... )
>>> store.ingest_table(df, schema)
>>> to_time_zone = ZoneInfo("Etc/GMT+5")
>>> dst_schema = store.localize_time_zone(
...     schema.name, to_time_zone, check_mapped_timestamps=True
... )

localize_time_zone_by_column(src_name: str, time_zone_column: str | None = None, output_file: Path | None = None, check_mapped_timestamps: bool = False) → TableSchema

Localize the time zone of the existing table represented by src_name to time zones defined by a column

Parameters:

• src_name – Refers to the table name of the source data.

• time_zone_column – Name of the time zone column for localization, default to None

• output_file – If set, write the mapped table to this Parquet file.

• check_mapped_timestamps – Perform time checks on the result of the mapping operation. This can be slow and is not required.

Raises:

TableAlreadyExists – Raised if the dst_schema name already exists.

Returns:

The schema of the newly created table.

Return type:

TableSchema

Examples

>>> store = Store()
>>> start = datetime(year=2018, month=1, day=1)  # tz-naive
>>> freq = timedelta(hours=1)
>>> hours_per_year = 8760
>>> num_time_arrays = 3
>>> df = pd.DataFrame(
...     {
...         "id": np.concatenate(
...             [np.repeat(i, hours_per_year) for i in range(1, 1 + num_time_arrays)]
...         ),
...         "timestamp": np.tile(
...             pd.date_range(start, periods=hours_per_year, freq="h"), num_time_arrays
...         ),
...         "time_zone": np.repeat(
...             ["Etc/GMT+5", "Etc/GMT+6", "Etc/GMT+7"], hours_per_year
...         ),  # EST, CST, MST
...         "value": np.random.random(hours_per_year * num_time_arrays),
...     }
... )
>>> schema = TableSchema(
...     name="some_data",
...     time_config=DatetimeRange(
...         time_column="timestamp",
...         start=start,
...         length=hours_per_year,
...         resolution=freq,
...     ),
...     time_array_id_columns=["id"],
...     value_column="value",
... )
>>> store.ingest_table(df, schema)
>>> time_zone_column = "time_zone"
>>> dst_schema = store.localize_time_zone_by_column(
...     schema.name,
...     time_zone_column,
...     check_mapped_timestamps=True,
... )

read_query(query: Table | str) → Table

Return the query result as an Ibis Table expression.

Call .execute() on the returned expression to materialize a pandas DataFrame.

Parameters:

query – SQL query as a string or ibis Table expression.

read_table(name: str) → Table

Return the table as an Ibis Table expression.

Call .execute() on the returned expression to materialize a pandas DataFrame.

read_raw_query(query: str, params: Any = None) → DataFrame

Execute a raw SQL query directly on the backend database connection, bypassing Ibis, and return the results as a DataFrame.

This is an escape hatch for executing backend-specific SQL that Ibis cannot express (e.g. SHOW TABLES, PRAGMA, SUMMARIZE). For portable queries, prefer read_query(), which returns an Ibis Table expression with consistent cross-backend typing.

Note: Unlike read_query(), no conversion of timestamps is performed. Timestamps will be in the format of the underlying database. SQLite backends will return strings instead of datetime.

Parameters:

• query – SQL query to execute.

• params – Optional parameters for the SQL query, in the backend’s native placeholder style (? for DuckDB and SQLite, named parameters for Spark).

Examples

>>> store = Store()
>>> df = store.read_raw_query(
...     "SELECT * from my_table WHERE column = ?", params=("value1",)
... )

write_query_to_parquet(stmt: Table | str, file_path: Path | str, overwrite: bool = False, partition_columns: list[str] | None = None) → None

Write the result of a query to a Parquet file.

Parameters:

• stmt – SQL query or ibis Table expression.

• file_path – Output Parquet file path.

• overwrite – Whether to overwrite an existing file.

• partition_columns – Optional columns to partition by.

write_table_to_parquet(name: str, file_path: Path | str, partition_columns: list[str] | None = None, overwrite: bool = False) → None

Write a table or view to a Parquet file.

delete_rows(name: str, time_array_id_values: dict[str, Any]) → int

Delete all rows matching the time_array_id_values.

Parameters:

• name – Name of table

• time_array_id_values – Values for the time_array_id_values. Keys must match the columns in the schema.

Returns:

Number of deleted rows

Return type:

int

drop_table(name: str, if_exists: bool = False) → None

Drop a table from the database.

create_view(schema: TableSchema, stmt: Table) → None

Create a view in the database.

drop_view(name: str, if_exists: bool = False) → None

Drop a view from the database.