TinyFlux API

See Getting Started to get TinyFlux up and running with writing and querying data.

Jump to an API section:

TinyFlux Database API

The main module of the TinyFlux package, containing the TinyFlux class.

class tinyflux.database.TinyFlux(*args, auto_index=True, **kwargs)

Bases: object

The TinyFlux class containing the interface for the TinyFlux package.

A facade singleton for the TinyFlux program. Manages the lifecycles of Storage, Index, and Measurement instances. Handles Points and Queries.

TinyFlux will reindex data in memory by default. To turn off this feature, set the value of ‘auto_index’ to false in the constructor keyword arguments.

TinyFlux will use the CSV store by default. To use a different store, pass a derived Storage subclass to the ‘storage’ keyword argument of the constructor.

All other args and kwargs are passed to the Storage instance.

Data Storage Model:

Data in TinyFlux is represented as Point objects. These are serialized and inserted into the TinyFlux storage layer in append-only fashion, providing the lowest-latency write op possible. This is of primary importance for time-series data which can often be written at a high- frequency. The schema of the storage layer is not rigid, allowing for variable metadata structures to be stored to the same data store.

Attributes:

storage: A reference to the Storage instance. index: A reference to the Index instance.

Usage:
>>> from tinyflux import TinyFlux
>>> db = TinyFlux("my_tf_db.csv")
all(sorted=True)

Get all data in the storage layer as Points.

Args:

sorted: Whether or not to return points sorted by time.

Returns:

A list of Points.

Return type:

List[Point]

close()

Close the database.

This may be needed if the storage instance used for this database needs to perform cleanup operations like closing file handles.

To ensure this method is called, the tinyflux instance can be used as a context manager:

>>> with TinyFlux('data.csv') as db:
        db.insert(Point())

Upon leaving this context, the ‘close’ method will be called.

Return type:

None

contains(query, measurement=None)

Check whether the database contains a point matching a query.

Defines a function that iterates over storage items and submits it to the storage layer.

Args:

query: A Query. measurement: An optional measurement to filter by.

Returns:

True if point found, else False.

Return type:

bool

count(query, measurement=None)

Count the points matching a query in the database.

Args:

query: a Query. measurement: An optional measurement to filter by.

Returns:

A count of matching points in the measurement.

Return type:

int

default_measurement_name = '_default'
default_storage_class

alias of CSVStorage

drop_measurement(name)

Drop a specific measurement from the database.

If ‘auto-index’ is True, a new index will be built.

Args:

name: The name of the measurement.

Returns:

The count of removed items.

Raises:

OSError if storage cannot be written to.

Return type:

int

get(query, measurement=None)

Get exactly one point specified by a query from the database.

Returns None if the point doesn’t exist.

Args:

query: A Query. measurement: An optional measurement to filter by.

Returns:

First found Point or None.

Return type:

Optional[Point]

get_field_keys(measurement=None)

Get all field keys in the database.

Args:

measurement: Optional measurement to filter by.

Returns:

List of field keys, sorted.

Return type:

List[str]

get_field_values(field_key, measurement=None)

Get field values in the database.

Args:

field_key: Field key to get values for. measurement: Optional measurement to filter by.

Returns:

List of field values.

Return type:

List[Union[int, float, str, bool, None]]

get_measurements()

Get the names of all measurements in the database.

Returns:

Names of all measurements in storage as a set.

Return type:

List[str]

get_tag_keys(measurement=None)

Get all tag keys in the database.

Args:

measurement: Optional measurement to filter by.

Returns:

List of field keys, sorted.

Return type:

List[str]

get_tag_values(tag_keys=[], measurement=None)

Get all tag values in the database.

Args:

tag_keys: Optional list of tag keys to get associated values for. measurement: Optional measurement to filter by.

Returns:

Mapping of tag_keys to associated tag values as a sorted list.

Return type:

Dict[str, List[Optional[str]]]

get_timestamps(measurement=None)

Get all timestamps in the database.

Returns timestamps in order of insertion in the database, as time-aware datetime objects with UTC timezone.

Args:

measurement: Optional measurement to filter by.

Returns:

List of timestamps by insertion order.

Return type:

List[datetime]

property index: Index

Get a reference to the index instance.

insert(point, measurement=None, compact_key_prefixes=False)

Insert a Point into the database.

Args:

point: A Point object. measurement: An optional measurement to filter by. compact_key_prefixes: Use compact key prefixes in relevant storages.

Returns:

1 if success.

Raises:

OSError if storage cannot be appended to. TypeError if point is not a Point instance.

Return type:

int

insert_multiple(points, measurement=None, compact_key_prefixes=False, batch_size=1000)

Insert Points into the database.

Points are processed in batches for memory efficiency when handling large datasets or generators. Each batch is written with one fsync.

Args:

points: An iterable of Point objects (can be generator/iterator). measurement: An optional measurement to insert Points into. compact_key_prefixes: Use compact key prefixes in relevant storages. batch_size: Number of points per batch (default: 1,000). Larger batches mean fewer fsync operations but more memory usage.

Returns:

The count of inserted points.

Raises:

OSError if storage cannot be appended to. TypeError if point is not a Point instance. ValueError if batch_size is less than 1.

Return type:

int

measurement(name, **kwargs)

Return a reference to a measurement in this database.

Chained methods will be handled by the Measurement class, and operate on the subset of Points belonging to the measurement.

A measurement does not need to exist in the storage layer for a Measurement object to be created.

Args:

name: Name of the measurement

Returns:

Reference to the measurement.

Return type:

Measurement

reindex()

Build a new in-memory index.

Raises:

OSError if storage cannot be written to.

Return type:

None

remove(query, measurement=None)

Remove Points from this database by query.

This is irreversible.

Args:

query: A query to remove Points by. measurement: An optional measurement to filter by.

Returns:

The count of removed points.

Raises:

OSError if storage cannot be written to.

Return type:

int

remove_all()

Remove all Points from this database.

This is irreversible.

Raises:

OSError if storage cannot be written to.

Return type:

None

search(query, measurement=None, sorted=True)

Get all points specified by a query.

Args:

query: A Query. measurement: An optional measurement to filter by. sorted: Whether or not to return the points sorted by time.

Returns:

A list of found Points.

Return type:

List[Point]

select(select_keys, query, measurement=None)

Get specified attributes from Points specified by a query.

‘select_keys’ should be an iterable of attributes including ‘time’, ‘measurement’, and tag keys and tag values. Passing ‘tags’ or ‘fields’ in the ‘select_keys’ iterable will not retrieve all tag and/or field values. Tag and field keys must be specified individually.

Args:

select_keys: A Point attribute or iterable of Point attributes. query: A Query. measurement: An optional measurement to filter by.

Returns:

A list of Point attribute values.

Return type:

List[Union[Any, Tuple[Any, ...]]]

property storage: Storage

Get a reference to the storage instance.

update(query, time=None, measurement=None, tags=None, fields=None, unset_fields=None, unset_tags=None, _measurement=None)

Update all matching Points in the database with new attributes.

Args:

query: A query as a condition. time: A datetime object or Callable returning one. measurement: A string or Callable returning one. tags: A mapping or Callable returning one. fields: A mapping or Callable returning one. unset_fields: Field keys to remove upon update. unset_tags: Tag keys to remove upon update. _measurement: An optional Measurement to filter by.

Returns:

A count of updated points.

Raises:

OSError if storage cannot be written to.

Return type:

int

update_all(time=None, measurement=None, tags=None, fields=None, unset_fields=None, unset_tags=None)

Update all points in the database with new attributes.

Args:

time: A datetime object or Callable returning one. measurement: A string or Callable returning one. tags: A mapping or Callable returning one. fields: A mapping or Callable returning one. unset_fields: Field keys to remove upon update. unset_tags: Tag keys to remove upon update.

Returns:

A count of updated points.

Raises:

OSError if storage cannot be written to.

Return type:

int

Point API

Definition of the TinyFlux Point class.

A Point is the data type upon which TinyFlux manages. It contains the time data and metadata for an individual observation. Points are serialized and deserialized from Storage. SimpleQuery act upon individual Points.

A Point is comprised of a timestamp, a measurement, fields, and tags.

Fields contains string/(numeric, string, or bool) key-values, while tags contain string/string key-values. This is enforced upon Point instantiation.

Usage:

>>> from tinyflux import Point
>>> p = Point(
        time=datetime.now(timezone.utc),
        measurement="my measurement",
        fields={"my field": 123.45},
        tags={"my tag key": "my tag value"}
    )
class tinyflux.point.Point(*args, **kwargs)

Bases: object

Define the Point class.

This is the only data type that TinyFlux handles directly. It is composed of a timestamp, measurement, tag-set, and field-set.

Usage:
>>> p = Point(
        time=datetime.now(timezone.utc),
        measurement="my measurement",
        fields={"my field": 123.45},
        tags={"my tag key": "my tag value"}
    )
default_measurement_name = '_default'
property fields: Dict[str, Union[int, float, str, bool, None]]

Get fields.

Setter accepts:

Any

property measurement: str

Get measurement.

Setter accepts:

Any

property tags: Dict[str, Optional[str]]

Get tags.

Setter accepts:

Any

property time: Optional[datetime]

Get time.

Setter accepts:

Any

tinyflux.point.validate_fields(fields)

Validate fields.

Args:

fields: The object to validate.

Raises:

ValueError: Exception if fields cannot be validated.

Return type:

None

tinyflux.point.validate_tags(tags)

Validate tags.

Args:

tags: The object to validate.

Raises:

ValueError: Exception if tags cannot be validated.

Return type:

None

Queries API

Definition of TinyFlux Queries.

A query contains logic in the form of a test and it acts upon a single Point when it is eventually evaluated.

All Queries begin as subclass of BaseQuery, which is not itself callable. Logic for the query is handled by the Python data model of the BaseQuery class, resulting in the generation of a SimpleQuery, which is callable. SimpleQuery instances support logical AND, OR, and NOT operations, which result in the initialization of a new CompoundQuery object.

Each SimpleQuery instance contains attributes that constitute the “deconstruction” of a query into several key parts (e.g. the operator, the right-hand side) so that the other consumers of queries, including an Index, may use them for their own purposes.

class tinyflux.queries.BaseQuery

Bases: object

A base class for the different TinyFlux query types.

A query type that explicitly unifies the divergent interfaces of TimeQuery, MeasurementQuery, TagQuery, and FieldQuery.

A BaseQuery is not itself callable. When it is combined with test logic, it generates a SimpleQuery, which is callable without exception.

Usage:
>>> from tinyflux import TagQuery, Point
>>> p = Point(tags={"city": "LA"})
>>> q1 = TagQuery()
>>> isinstance(q1, tinyflux.queries.BaseQuery)
True
>>> q1(p)
RuntimeError: Empty query was evaluated.
>>> q2 = TagQuery().city == "LA"
>>> q2
SimpleQuery('tags', '==', ('city',), 'LA')
>>> q2(p)
True
is_hashable()

Return hash is not empty.

Return type:

bool

map(func)

Add a function to the query path.

Similar to __getattr__ but for arbitrary functions.

Args:

func: The function to add.

Return type:

BaseQuery

matches(regex, flags=0)

Run a regex test against a value (whole string has to match).

>>> TagQuery().f1.matches(r'^\\w+$')
Args:

regex: The regular expression to use for matching. flags: Regex flags to pass to re.match.

Return type:

SimpleQuery

noop()

Evaluate to True.

Useful for having a base value when composing queries dynamically.

Return type:

SimpleQuery

search(regex, flags=0)

Run a regex test against a value (only substring has to match).

>>> TagQuery().f1.search(r'^\\w+$')
Args:

regex: The regular expression to use for matching. flags: Regex flags to pass to re.match.

Return type:

SimpleQuery

test(func, *args)

Run a user-defined test function against a value.

>>> def test_func(val):
...     return val == 42
...
>>> FieldQuery()["my field"].test(test_func)
Warning:

The test function provided needs to be deterministic (returning the same value when provided with the same arguments), otherwise this may mess up the query cache that Table implements.

Args:

func: The function to call, passing the value as the first arg. args: Additional arguments to pass to the test function.

Return type:

SimpleQuery

class tinyflux.queries.CompoundQuery(query1, query2, operator, hashval)

Bases: object

A container class for simple and/or compound queries and an operator.

A CompoundQuery is generated by built-in __and__, __or__, and __not__ operations on a SimpleQuery.

Attributes:

query1: A SimpleQuery or CompoundQuery instance. query2: A SimpleQuery or CompoundQuery instance. operator: The operator.

Usage:
>>> from tinyflux import FieldQuery, TagQuery
>>> time_q = FieldQuery().temp_f < 55.0
>>> tags_q = TagQuery().city == "Los Angeles"
>>> cold_LA_q = time_q & tags_q
>>> type(cold_LA_q)
<class 'tinyflux.queries.CompoundQuery'>
is_hashable()

Return the ability to hash this query.

Return type:

bool

class tinyflux.queries.FieldQuery

Bases: BaseQuery

The base query for Point fields.

Generates a SimpleQuery that evaluates Point ‘fields’ attributes.

Usage:
>>> from tinyflux import FieldQuery
>>> my_field_q = FieldQuery().my_field == 10.0
exists()

Test at Point where a provided key exists.

Usage:
>>> FieldQuery().my_field.exists()
Return type:

SimpleQuery

matches(regex, flags=0)

Raise an exception for regex query.

Return type:

SimpleQuery

search(regex, flags=0)

Raise an exception for regex query.

Return type:

SimpleQuery

class tinyflux.queries.MeasurementQuery

Bases: BaseQuery

The base query for Point measurement.

Generates a SimpleQuery that evaluates Point ‘measurement’ attributes.

Usage:
>>> from tinyflux import MeasurementQuery
>>> my_measurement_q = MeasurementQuery() == "my measurement"
class tinyflux.queries.SimpleQuery(point_attr, operator, rhs, test, path_resolver, path, hashval)

Bases: object

A single query instance.

This is the object on which the actual query operations are performed. The BaseQuery class acts like a query builder and generates SimpleQuery objects which will evaluate their query against a given point when called.

Query instances can be combined using logical OR and AND and inverted using logical NOT.

A SimpleQuery can be parsed using private attributes.

TODO: In order to be usable in a query cache, a query needs to have a stable hash value with the same query always returning the same hash. That way a query instance can be used as a key in a dictionary.

Usage:
>>> from tinyflux import TagQuery
>>> los_angeles_q = TagQuery().city == "Los Angeles"
>>> type(los_angeles_q)
<class 'tinyflux.queries.SimpleQuery'>
is_hashable()

Return the ability to hash this query.

Return type:

bool

property point_attr: str

Get the attribute of a Point object relevant for this query.

class tinyflux.queries.TagQuery

Bases: BaseQuery

The base query for Point tags.

Generates a SimpleQuery that evaluates Point ‘tags’ attributes.

Usage:
>>> from tinyflux import TagQuery
>>> my_tag_q = TagQuery().my_tag_key == "my tag value"
exists()

Test a Point where a provided key exists.

>>> TagQuery().my_tag.exists()
Return type:

SimpleQuery

class tinyflux.queries.TimeQuery

Bases: BaseQuery

The base query for Point time.

Generates a SimpleQuery that evaluates Point ‘measurement’ attributes.

Usage:
>>> from datetime import datetime, timezone
>>> from tinyflux import TimeQuery
>>> my_time_q = TimeQuery() < datetime.now(timezone.utc)
matches(regex, flags=0)

Raise an exception for regex query.

Return type:

SimpleQuery

search(regex, flags=0)

Raise an exception for regex query.

Return type:

SimpleQuery

Measurement API

Definition of TinyFlux measurement class.

The measurement class provides a convenient interface into a subset of data points with a common measurement name. A measurement is analogous to a table in a traditional RDBMS.

Usage:
>>> db = TinyFlux(storage=MemoryStorage)
>>> m = db.measurement("my_measurement")
class tinyflux.measurement.Measurement(name, db)

Bases: object

Define the Measurement class.

Measurement objects are created at runtime when the TinyFlux ‘measurement’ method is invoked.

Attributes:

name: Name of the measurement. storage: Storage object for the measurement’s parent TinyFlux db. index: Index object for the measurement’s parent TinyFlux db.

all(sorted=True)

Get all points in this measurement.

Args:

sorted: Whether or not to return points in sorted time order.

Returns:

A list of points.

Return type:

List[Point]

contains(query)

Check whether the measurement contains a point matching a query.

Args:

query: A SimpleQuery.

Returns:

True if point found, else False.

Return type:

bool

count(query)

Count the points matching a query in this measurement.

Args:

query: a SimpleQuery.

Returns:

A count of matching points in the measurement.

Return type:

int

get(query)

Get exactly one point specified by a query from this measurement.

Returns None if the point doesn’t exist.

Args:

query: A SimpleQuery.

Returns:

First found Point or None.

Return type:

Optional[Point]

get_field_keys()

Get all field keys for this measurement.

Returns:

List of field keys, sorted.

Return type:

List[str]

get_field_values(field_key)

Get field values from this measurement for the specified key.

Args:

field_key: The field key to get field values for.

Returns:

List of field keys, sorted.

Return type:

List[Union[int, float, str, bool, None]]

get_tag_keys()

Get all tag keys for this measurement.

Returns:

List of tag keys, sorted.

Return type:

List[str]

get_tag_values(tag_keys=[])

Get all tag values in the database.

Args:

tag_keys: Optional list of tag keys to get associated values for.

Returns:

Mapping of tag_keys to associated tag values as a sorted list.

Return type:

Dict[str, List[str]]

get_timestamps()

Get all timestamps in the database.

Returns timestamps in order of insertion in the database, as time-aware datetime objects with UTC timezone.

Args:

measurement: Optional measurement to filter by.

Returns:

List of timestamps by insertion order.

Return type:

List[datetime]

property index: Index

Get the measurement storage instance.

insert(point)

Insert a Point into a measurement.

If the passed Point has a different measurement value, ‘insert’ will update the measurement value with that of this measurement.

Args:

point: A Point object.

Returns:

1 if success.

Raises:

TypeError if point is not a Point instance.

Return type:

int

insert_multiple(points, batch_size=1000)

Insert Points into this measurement.

If the passed Point has a different measurement value, ‘insert’ will update the measurement value with that of this measurement.

Points are processed in batches for memory efficiency when handling large datasets or generators. Each batch is written with one fsync.

Args:

points: An iterable of Point objects (can be generator/iterator). batch_size: Number of points per batch (default: 1,000). Larger batches mean fewer fsync operations but more memory usage.

Returns:

The count of inserted points.

Raises:

TypeError if point is not a Point instance. ValueError if batch_size is less than 1.

Return type:

int

property name: str

Get the measurement name.

remove(query)

Remove Points from this measurement by query.

This is irreversible.

Returns:

The count of removed points.

Return type:

int

remove_all()

Remove all Points from this measurement.

This is irreversible.

Returns:

The count of removed points.

Return type:

int

search(query, sorted=True)

Get all points specified by a query from this measurement.

Order is not guaranteed. Returns empty list if no points are found.

Args:

query: A SimpleQuery. sorted: Whether or not to return points sorted by timestamp.

Returns:

A list of found Points.

Return type:

List[Point]

select(keys, query)

Get specified attributes from Points specified by a query.

‘keys’ should be an iterable of attributes including ‘time’, ‘measurement’, and tag keys and tag values. Passing ‘tags’ or ‘fields’ in the ‘keys’ iterable will not retrieve all tag and/or field values. Tag and field keys must be specified individually.

Args:

keys: An iterable of Point attributes. query: A Query.

Returns:

A list of tuples of Point attribute values.

Return type:

List[Tuple[Union[datetime, str, int, float, None]]]

property storage: Storage

Get the measurement storage instance.

update(query, time=None, measurement=None, tags=None, fields=None, unset_fields=None, unset_tags=None)

Update all matching Points in this measurement with new attributes.

Args:

query: A query. time: A datetime object or Callable returning one. measurement: A string or Callable returning one. tags: A mapping or Callable returning one. fields: A mapping or Callable returning one. unset_fields: Field keys to remove upon update. unset_tags: Tag keys to remove upon update.

Returns:

A count of updated points.

Return type:

int

update_all(time=None, measurement=None, tags=None, fields=None, unset_fields=None, unset_tags=None)

Update all matching Points in this measurement with new attributes.

Args:

query: A query. time: A datetime object or Callable returning one. measurement: A string or Callable returning one. tags: A mapping or Callable returning one. fields: A mapping or Callable returning one. unset_fields: Field keys to remove upon update. unset_tags: Tag keys to remove upon update.

Returns:

A count of updated points.

Return type:

int

Index API

Definition of the TinyFlux Index.

Class descriptions for Index and IndexResult. An Index acts like a singleton, and is initialized at creation time with the TinyFlux instance. It provides efficient in-memory data structures and getters for TinyFlux operations. An Index instance is not a part of the TinyFlux interface.

An IndexResult returns the indices of relevant TinyFlux queries for further handling, usually as an input to a storage retrieval.

class tinyflux.index.Index(valid=True)

Bases: object

An in-memory index for the storage instance.

Provides efficient data structures and searches for TinyFlux data. An Index instance is created and its lifetime is handled by a TinyFlux instance.

Attributes:

empty: Index contains no items (used in testing). valid: Index represents current state of TinyFlux.

build(points)

Build the index from scratch.

Args:

points: The collection of points to build the Index from.

Usage:
>>> i = Index().build([Point()])
Return type:

None

property empty: bool

Return True if index is empty.

get_field_keys(measurement=None)

Get field keys from this index, optionally filtered by measurement.

Args:

measurement: Optional measurement to filter by.

Returns:

Set of field keys.

Return type:

Set[str]

get_field_values(field_key, measurement=None)

Get field values from this index, optionally filter by measurement.

Args:

field_key: Field key to get field values for. measurement: Optional measurement to filter by.

Returns:

List of field values.

Return type:

List[Union[int, float, str, bool, None]]

get_measurements()

Get the names of all measurements in the Index.

Returns:

Unique names of measurements as a set.

Usage:
>>> n = Index().build([Point()]).get_measurements()
Return type:

Set[str]

get_tag_keys(measurement=None)

Get tag keys from this index, optionally filtered by measurement.

Args:

measurement: Optional measurement to filter by.

Returns:

Set of field keys.

Return type:

Set[str]

get_tag_values(tag_keys=[], measurement=None)

Get all tag values from the index.

Args:

tag_keys: Optional list of tag keys to get associated values for. measurement: Optional measurement to filter by.

Returns:

Mapping of tag_keys to associated tag values as a set.

Return type:

Dict[str, Set[Optional[str]]]

get_timestamps(measurement=None)

Get timestamps from the index.

Args:

measurement: Optional measurement to filter by.

Returns:

List of timestamps.

Return type:

List[float]

insert(points=[])

Update index with new points.

Accepts new points to add to an Index. Points are assumed to be passed to this method in non-descending time order.

Args:

points: List of tinyflux.Point instances.

Usage:
>>> Index().insert([Point()])
Return type:

None

invalidate()

Invalidate an Index.

This method is invoked when the Index no longer represents the current state of TinyFlux and its Storage instance.

Usage:
>>> i = Index()
>>> i.invalidate()
Return type:

None

property latest_time: datetime

Return the latest time in the index.

remove(r_items)

Remove items from the index.

Return type:

None

search(query)

Handle a TinyFlux query.

Parses the query, generates a new IndexResult, and returns it.

Args:

query: A tinyflux.queries.Query.

Returns:

An IndexResult instance.

Usage:
>>> i = Index().build([Point()])
>>> q = TimeQuery() < datetime.now(timezone.utc)
>>> r = i.search(q)
Return type:

IndexResult

update(u_items)

Update the index.

Args:

u_items: A mapping of old indices to update indices.

Return type:

None

property valid: bool

Return an empty index.

class tinyflux.index.IndexResult(items, index_count)

Bases: object

Returns indices of TinyFlux queries that are handled by an Index.

IndexResults instances are generated by an Index.

Attributes:

items: A set of indices as ints.

Usage:
>>> IndexResult(items=set(), index_count=0)
property items: Set[int]

Return query result items.

Storages API

Definition of TinyFlux storages classes.

Storage defines an abstract base case using the built-in ABC of python. This class defines the requires abstract methods of read, write, and append, as well as getters and setters for attributes required to reindex the data.

A storage object will manage data with a file handle, or in memory.

A storage class is provided to the TinyFlux facade as an initial argument. The TinyFlux instance will manage the lifecycle of the storage instance.

Usage:
>>> my_mem_db = TinyFlux(storage=MemoryStorage)
>>> my_csv_db = TinyFlux('path/to/my.csv', storage=CSVStorage)
class tinyflux.storages.CSVStorage(path, create_dirs=False, encoding=None, access_mode='r+', flush_on_insert=True, newline='', **kwargs)

Bases: Storage

Define the default storage instance for TinyFlux, a CSV store.

CSV provides append-only writes, which is efficient for high-frequency writes, common to time-series datasets.

Usage:
>>> from tinyflux import CSVStorage
>>> db = TinyFlux("my_csv_store.csv", storage=CSVStorage)
append(items, temporary=False)

Append points to the CSV store.

Args:

items: A list of objects. temporary: Whether or not to append to temporary storage.

Return type:

None

property can_append: bool

Return whether or not appends can occur.

property can_read: bool

Return whether or not reads can occur.

property can_write: bool

Return whether or not writes can occur.

close()

Clean up data store.

Closes the file object.

Return type:

None

read()

Read all items from the storage into memory.

Returns:

A list of Point objects.

Return type:

List[Point]

reset()

Reset the storage instance.

Removes all data.

Return type:

None

class tinyflux.storages.MemoryStorage

Bases: Storage

Define the in-memory storage instance for TinyFlux.

Memory is cleaned up along with the parent process.

Attributes:

_initially_empty: No data in the storage instance. _memory: List of Points. _temp_memory: List of Points.

Usage:
>>> from tinyflux import MemoryStorage
>>> db = TinyFlux(storage=MemoryStorage)
append(items, temporary=False)

Append points to the memory.

Args:

points: A list of Point objects. temporary: Whether or not to append to temporary storage.

Return type:

None

read()

Read data from the store.

Returns:

A list of Point objects.

Return type:

List[Point]

reset()

Reset the storage instance.

Removes all data.

Return type:

None

class tinyflux.storages.Storage

Bases: ABC

The abstract base class for all storage types for TinyFlux.

Defines an extensible, static interface with required read/write ops and index-related getter/setters.

Custom storage classes should inherit like so:
>>> from tinyflux import Storage
>>> class MyStorageClass(Storage):
        ...
abstractmethod append(points, temporary=False)

Append points to the store.

Args:

points: A list of Point objects. temporary: Whether or not to append to temporary storage.

Return type:

None

property can_append: bool

Can append to DB.

property can_read: bool

Can read the DB.

property can_write: bool

Can write to DB.

close()

Perform clean up ops.

Return type:

None

abstractmethod read()

Read from the store.

Re-ordering the data after a read provides TinyFlux with the ability to build an index.

Args:

reindex_on_read: Reorder the store after data is read.

Returns:

A list of Points.

Return type:

List[Point]

abstractmethod reset()

Reset the storage instance.

Removes all data.

Return type:

None

tinyflux.storages.create_file(path, create_dirs)

Create a file if it doesn’t exist yet.

Args:

path: The file to create. create_dirs: Whether to create all missing parent directories.

Return type:

None

Utils API

Definition of TinyFlux utils.

class tinyflux.utils.FrozenDict

Bases: dict

An immutable dictionary.

This is used to generate stable hashes for queries that contain dicts. Usually, Python dicts are not hashable because they are mutable. This class removes the mutability and implements the __hash__ method.

From TinyDB.

clear(*args, **kwargs)

Raise a TypeError for a given dict method.

Return type:

None

pop(k, d=None)

Raise TypeError for pop.

Return type:

None

popitem(*args, **kwargs)

Raise a TypeError for a given dict method.

Return type:

None

update(*args, **kwargs)

Raise TypeError for update.

Return type:

None

tinyflux.utils.find_eq(sorted_list, x)

Locate the leftmost value exactly equal to x.

Args:

sorted_list: The list to search. x: The element to search.

Returns:

The index of the found element or None.

Return type:

Optional[int]

tinyflux.utils.find_ge(sorted_list, x)

Find leftmost item greater than or equal to x.

Args:

sorted_list: The list to search. x: The element to search.

Returns:

The index of the found element or None.

Return type:

Optional[int]

tinyflux.utils.find_gt(sorted_list, x)

Find leftmost value greater than x.

Args:

sorted_list: The list to search. x: The element to search.

Returns:

The index of the found element or None.

Return type:

Optional[int]

tinyflux.utils.find_le(sorted_list, x)

Find rightmost value less than or equal to x.

Args:

sorted_list: The list to search. x: The element to search.

Returns:

The index of the found element or None.

Return type:

Optional[int]

tinyflux.utils.find_lt(sorted_list, x)

Find rightmost value less than x.

Args:

sorted_list: The list to search. x: The element to search.

Returns:

The index of the found element or None.

Return type:

Optional[int]

tinyflux.utils.freeze(obj)

Freeze an object by making it immutable and thus hashable.

Args:

obj: Any python object.

Returns:

The object in a hashable form.

Return type:

object