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.

Return type:

List[Point]

Args:

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

Returns:

A list of Points.

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:

Return type:

None

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

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

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.

Return type:

bool

Args:

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

Returns:

True if point found, else False.

count(query, measurement=None)

Count the points matching a query in the database.

Return type:

int

Args:

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

Returns:

A count of matching points in the measurement.

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.

Return type:

int

Args:

name: The name of the measurement.

Returns:

The count of removed items.

Raises:

OSError if storage cannot be written to.

get(query, measurement=None)

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

Returns None if the point doesn’t exist.

Return type:

Optional[Point]

Args:

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

Returns:

First found Point or None.

get_field_keys(measurement=None)

Get all field keys in the database.

Return type:

List[str]

Args:

measurement: Optional measurement to filter by.

Returns:

List of field keys, sorted.

get_field_values(field_key, measurement=None)

Get field values in the database.

Return type:

List[Union[int, float, None]]

Args:

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

Returns:

List of field values.

get_measurements()

Get the names of all measurements in the database.

Return type:

List[str]

Returns:

Names of all measurements in storage as a set.

get_tag_keys(measurement=None)

Get all tag keys in the database.

Return type:

List[str]

Args:

measurement: Optional measurement to filter by.

Returns:

List of field keys, sorted.

get_tag_values(tag_keys=[], measurement=None)

Get all tag values in the database.

Return type:

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

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.

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.

Return type:

List[datetime]

Args:

measurement: Optional measurement to filter by.

Returns:

List of timestamps by insertion order.

property index: Index

Get a reference to the index instance.

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

Insert a Point into the database.

Return type:

int

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 appendex to. TypeError if point is not a Point instance.

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

Insert Points into the database.

Return type:

int

Args:

points: An iterable of Point objects. measurement: An optional measurement to insert Points into. compact_key_prefixes: Use compact key prefixes in relevant storages.

Returns:

The count of inserted points.

Raises:

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

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.

Return type:

Measurement

Args:

name: Name of the measurement

Returns:

Reference to the measurement.

reindex()

Build a new in-memory index.

Return type:

None

Raises:

OSError if storage cannot be written to.

remove(query, measurement=None)

Remove Points from this database by query.

This is irreversible.

Return type:

int

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.

remove_all()

Remove all Points from this database.

This is irreversible.

Return type:

None

Raises:

OSError if storage cannot be written to.

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

Get all points specified by a query.

Return type:

List[Point]

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.

select(select_keys, query, measurement=None)

Get specified attributes from Points specified by a query.

‘select_keys’ should be an iterable of attributres 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.

Return type:

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

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.

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.

Return type:

int

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.

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.

Return type:

int

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.

tinyflux.database.append_op(method)

Decorate an append operation with assertion.

Ensures storage can be appended to before doing anything.

Return type:

Callable[..., Any]

tinyflux.database.read_op(method)

Decorate a read operation with assertion.

Ensures storage can be read from before doing anything.

Return type:

Callable[..., Any]

tinyflux.database.temp_storage_op(method)

Decorate a db operation that requires auxiliary storage.

Initializes temporary storage, invokes method, and cleans-up storage after op has run.

Return type:

Callable[..., Any]

tinyflux.database.write_op(method)

Decorate a write operation with assertion.

Ensures storage can be written to before doing anything.

Return type:

Callable[..., Any]

Point API

Defintion 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. SimpleQuerys act upon individual Points.

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

Fields contains string/numeric 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, int | float | None]

Get fields.

property measurement: str

Get measurement.

property tags: Dict[str, str | None]

Get tags.

property time: datetime | None

Get time.

tinyflux.point.validate_fields(fields)

Validate fields.

Return type:

None

Args:

fields: The object to validate.

Raises:

ValueError: Exception if fields cannot be validated.

tinyflux.point.validate_tags(tags)

Validate tags.

Return type:

None

Args:

tags: The object to validate.

Raises:

ValueError: Exception if tags cannot be validated.

Queries API

Defintion 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 “deconstuction” of a query into several key parts (e.g. the operator, the right-hand side) so that the other consumers of queries, includng 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 explicity unifies the divergent interfaces of TimeQuery, MeasurementQuery, TagQuery, and FieldQuery.

A BaseQuery is not iteslf 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.

Return type:

BaseQuery

Args:

func: The function to add.

matches(regex, flags=0)

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

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

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

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). :rtype: SimpleQuery

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

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

test(func, *args)

Run a user-defined test function against a value.

>>> def test_func(val):
...     return val == 42
:rtype: :py:class:`~tinyflux.queries.SimpleQuery`

… >>> FieldQuery()[“my field”].test(test_func)

Warning:

The test fuction 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.

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.

Return type:

SimpleQuery

Usage:
>>> FieldQuery().my_field.exists()
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, 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. :rtype: SimpleQuery

>>> TagQuery().my_tag.exists()
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

Defintion 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.

Return type:

List[Point]

Args:

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

Returns:

A list of points.

contains(query)

Check whether the measurement contains a point matching a query.

Return type:

bool

Args:

query: A SimpleQuery.

Returns:

True if point found, else False.

count(query)

Count the points matching a query in this measurement.

Return type:

int

Args:

query: a SimpleQuery.

Returns:

A count of matching points in the measurement.

get(query)

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

Returns None if the point doesn’t exist.

Return type:

Optional[Point]

Args:

query: A SimpleQuery.

Returns:

First found Point or None.

get_field_keys()

Get all field keys for this measurement.

Return type:

List[str]

Returns:

List of field keys, sorted.

get_field_values(field_key)

Get field values from this measurement for the specified key.

Return type:

List[Union[int, float, None]]

Args:

field_key: The field key to get field values for.

Returns:

List of field keys, sorted.

get_tag_keys()

Get all tag keys for this measurement.

Return type:

List[str]

Returns:

List of tag keys, sorted.

get_tag_values(tag_keys=[])

Get all tag values in the database.

Return type:

Dict[str, List[str]]

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.

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.

Return type:

List[datetime]

Args:

measurement: Optional measurement to filter by.

Returns:

List of timestamps by insertion order.

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.

Return type:

int

Args:

point: A Point object.

Returns:

1 if success.

Raises:

TypeError if point is not a Point instance.

insert_multiple(points)

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.

Return type:

int

Args:

points: An iterable of Point objects.

Returns:

The count of inserted points.

Raises:

TypeError if point is not a Point instance.

property name: str

Get the measurement name.

remove(query)

Remove Points from this measurement by query.

This is irreversible.

Return type:

int

Returns:

The count of removed points.

remove_all()

Remove all Points from this measurement.

This is irreversible.

Return type:

int

Returns:

The count of removed points.

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.

Return type:

List[Point]

Args:

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

Returns:

A list of found Points.

select(keys, query)

Get specified attributes from Points specified by a query.

‘keys’ should be an iterable of attributres 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.

Return type:

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

Args:

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

Returns:

A list of tuples of Point attribute values.

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.

Return type:

int

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.

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.

Return type:

int

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.

Index API

Defintion 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 indicies of revelant 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.

Return type:

None

Args:

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

Usage:
>>> i = Index().build([Point()])
property empty: bool

Return True if index is empty.

get_field_keys(measurement=None)

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

Return type:

Set[str]

Args:

measurement: Optional measurement to filter by.

Returns:

Set of field keys.

get_field_values(field_key, measurement=None)

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

Return type:

List[Union[int, float, None]]

Args:

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

Returns:

List of field values.

get_measurements()

Get the names of all measurements in the Index.

Return type:

Set[str]

Returns:

Unique names of measurements as a set.

Usage:
>>> n = Index().build([Point()]).get_measurements()
get_tag_keys(measurement=None)

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

Return type:

Set[str]

Args:

measurement: Optional measurement to filter by.

Returns:

Set of field keys.

get_tag_values(tag_keys=[], measurement=None)

Get all tag values from the index.

Return type:

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

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.

get_timestamps(measurement=None)

Get timestamps from the index.

Return type:

List[float]

Args:

measurement: Optional measurement to filter by.

Returns:

List of timestamps.

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.

Return type:

None

Args:

points: List of tinyflux.Point instances.

Usage:
>>> Index().insert([Point()])
invalidate()

Invalidate an Index.

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

Return type:

None

Usage:
>>> i = Index()
>>> i.invalidate()
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.

Return type:

IndexResult

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)
update(u_items)

Update the index.

Return type:

None

Args:

u_items: A mapping of old indices to update indices.

property valid: bool

Return an empty index.

class tinyflux.index.IndexResult(items, index_count)

Bases: object

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

IndexResults instances are generated by an Index.

Arritributes:

items: A set of indicies as ints.

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

Return query result items.

Storages API

Defintion 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.

Return type:

None

Args:

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

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.

Return type:

List[Point]

Returns:

A list of Point objects.

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.

Return type:

None

Args:

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

read()

Read data from the store.

Return type:

List[Point]

Returns:

A list of Point objects.

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 inheret like so:
>>> from tinyflux import Storage
>>> class MyStorageClass(Storage):
        ...
abstract append(points, temporary=False)

Append points to the store.

Return type:

None

Args:

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

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

abstract read()

Read from the store.

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

Return type:

List[Point]

Args:

reindex_on_read: Reorder the store after data is read.

Returns:

A list of Points.

abstract 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.

Return type:

None

Args:

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

Utils API

Defintion 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.

Return type:

Optional[int]

Args:

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

Returns:

The index of the found element or None.

tinyflux.utils.find_ge(sorted_list, x)

Find leftmost item greater than or equal to x.

Return type:

Optional[int]

Args:

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

Returns:

The index of the found element or None.

tinyflux.utils.find_gt(sorted_list, x)

Find leftmost value greater than x.

Return type:

Optional[int]

Args:

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

Returns:

The index of the found element or None.

tinyflux.utils.find_le(sorted_list, x)

Find rightmost value less than or equal to x.

Return type:

Optional[int]

Args:

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

Returns:

The index of the found element or None.

tinyflux.utils.find_lt(sorted_list, x)

Find rightmost value less than x.

Return type:

Optional[int]

Args:

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

Returns:

The index of the found element or None.

tinyflux.utils.freeze(obj)

Freeze an object by making it immutable and thus hashable.

Return type:

object

Args:

obj: Any python object.

Returns:

The object in a hashable form.