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 tinyflux.storages.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, 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[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: tinyflux.index.Index

Get a reference to the index instance.

Return type

Index

insert(point, measurement=None)

Insert a Point into the database.

Args:

point: A Point object. measurement: An optional measurement to filter by.

Returns:

1 if success.

Raises:

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

Return type

int

insert_multiple(points, measurement=None)

Insert Points into the database.

Args:

points: An iterable of Point objects. measurement: An optional measurement to insert Points into.

Returns:

The count of inserted points.

Raises:

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

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

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: tinyflux.storages.Storage

Get a reference to the storage instance.

Return type

Storage

update(query, time=None, measurement=None, tags=None, fields=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. _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)

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.

Returns:

A count of updated points.

Raises:

OSError if storage cannot be written to.

Return type

int

tinyflux.database.append_op(method)

Decorate an append operation with assertion.

Ensures storage can be appended to before doing anything.

tinyflux.database.read_op(method)

Decorate a read operation with assertion.

Ensures storage can be read from before doing anything.

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.

tinyflux.database.write_op(method)

Decorate a write operation with assertion.

Ensures storage can be written to before doing anything.


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

Get fields.

property measurement

Get measurement.

property tags

Get tags.

property time

Get time.

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

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.

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

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: tinyflux.queries.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: tinyflux.queries.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.

Return type

str

class tinyflux.queries.TagQuery

Bases: tinyflux.queries.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: tinyflux.queries.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.

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, 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: tinyflux.index.Index

Get the measurement storage instance.

Return type

Index

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)

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.

Args:

points: An iterable of Point objects.

Returns:

The count of inserted points.

Raises:

TypeError if point is not a Point instance.

Return type

int

property name: str

Get the measurement name.

Return type

str

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

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: tinyflux.storages.Storage

Get the measurement storage instance.

Return type

Storage

update(query, time=None, measurement=None, tags=None, fields=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.

Returns:

A count of updated points.

Return type

int

update_all(time=None, measurement=None, tags=None, fields=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.

Returns:

A count of updated points.

Return type

int


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.

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.

Return type

bool

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, 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[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 lateset_time: datetime.datetime

Return the lastest time in the index.

Return type

datetime

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.

Return type

bool

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.

Return type

Set[int]


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: tinyflux.storages.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.

Return type

bool

property can_read: bool

Return whether or not reads can occur.

Return type

bool

property can_write: bool

Return whether or not writes can occur.

Return type

bool

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: tinyflux.storages.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.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.

Args:

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

Return type

None

property can_append: bool

Can append to DB.

Return type

bool

property can_read: bool

Can read the DB.

Return type

bool

property can_write: bool

Can write to DB.

Return type

bool

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.

Args:

reindex_on_read: Reorder the store after data is read.

Returns:

A list of Points.

Return type

List[Point]

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.

Args:

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

Return type

None


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, **kws)

Raise a TypeError for a given dict method.

pop(k, d=None)

Raise TypeError for pop.

popitem(*args, **kws)

Raise a TypeError for a given dict method.

update(e=None, **f)

Raise TypeError for update.

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