| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896 |
- from typing import Dict, List, Optional, Tuple, Union
- from redis.exceptions import DataError
- from redis.typing import KeyT, Number
- ADD_CMD = "TS.ADD"
- ALTER_CMD = "TS.ALTER"
- CREATERULE_CMD = "TS.CREATERULE"
- CREATE_CMD = "TS.CREATE"
- DECRBY_CMD = "TS.DECRBY"
- DELETERULE_CMD = "TS.DELETERULE"
- DEL_CMD = "TS.DEL"
- GET_CMD = "TS.GET"
- INCRBY_CMD = "TS.INCRBY"
- INFO_CMD = "TS.INFO"
- MADD_CMD = "TS.MADD"
- MGET_CMD = "TS.MGET"
- MRANGE_CMD = "TS.MRANGE"
- MREVRANGE_CMD = "TS.MREVRANGE"
- QUERYINDEX_CMD = "TS.QUERYINDEX"
- RANGE_CMD = "TS.RANGE"
- REVRANGE_CMD = "TS.REVRANGE"
- class TimeSeriesCommands:
- """RedisTimeSeries Commands."""
- def create(
- self,
- key: KeyT,
- retention_msecs: Optional[int] = None,
- uncompressed: Optional[bool] = False,
- labels: Optional[Dict[str, str]] = None,
- chunk_size: Optional[int] = None,
- duplicate_policy: Optional[str] = None,
- ):
- """
- Create a new time-series.
- Args:
- key:
- time-series key
- retention_msecs:
- Maximum age for samples compared to highest reported timestamp (in milliseconds).
- If None or 0 is passed then the series is not trimmed at all.
- uncompressed:
- Changes data storage from compressed (by default) to uncompressed
- labels:
- Set of label-value pairs that represent metadata labels of the key.
- chunk_size:
- Memory size, in bytes, allocated for each data chunk.
- Must be a multiple of 8 in the range [128 .. 1048576].
- duplicate_policy:
- Policy for handling multiple samples with identical timestamps.
- Can be one of:
- - 'block': an error will occur for any out of order sample.
- - 'first': ignore the new value.
- - 'last': override with latest value.
- - 'min': only override if the value is lower than the existing value.
- - 'max': only override if the value is higher than the existing value.
- For more information: https://redis.io/commands/ts.create/
- """ # noqa
- params = [key]
- self._append_retention(params, retention_msecs)
- self._append_uncompressed(params, uncompressed)
- self._append_chunk_size(params, chunk_size)
- self._append_duplicate_policy(params, CREATE_CMD, duplicate_policy)
- self._append_labels(params, labels)
- return self.execute_command(CREATE_CMD, *params)
- def alter(
- self,
- key: KeyT,
- retention_msecs: Optional[int] = None,
- labels: Optional[Dict[str, str]] = None,
- chunk_size: Optional[int] = None,
- duplicate_policy: Optional[str] = None,
- ):
- """
- Update the retention, chunk size, duplicate policy, and labels of an existing
- time series.
- Args:
- key:
- time-series key
- retention_msecs:
- Maximum retention period, compared to maximal existing timestamp (in milliseconds).
- If None or 0 is passed then the series is not trimmed at all.
- labels:
- Set of label-value pairs that represent metadata labels of the key.
- chunk_size:
- Memory size, in bytes, allocated for each data chunk.
- Must be a multiple of 8 in the range [128 .. 1048576].
- duplicate_policy:
- Policy for handling multiple samples with identical timestamps.
- Can be one of:
- - 'block': an error will occur for any out of order sample.
- - 'first': ignore the new value.
- - 'last': override with latest value.
- - 'min': only override if the value is lower than the existing value.
- - 'max': only override if the value is higher than the existing value.
- For more information: https://redis.io/commands/ts.alter/
- """ # noqa
- params = [key]
- self._append_retention(params, retention_msecs)
- self._append_chunk_size(params, chunk_size)
- self._append_duplicate_policy(params, ALTER_CMD, duplicate_policy)
- self._append_labels(params, labels)
- return self.execute_command(ALTER_CMD, *params)
- def add(
- self,
- key: KeyT,
- timestamp: Union[int, str],
- value: Number,
- retention_msecs: Optional[int] = None,
- uncompressed: Optional[bool] = False,
- labels: Optional[Dict[str, str]] = None,
- chunk_size: Optional[int] = None,
- duplicate_policy: Optional[str] = None,
- ):
- """
- Append (or create and append) a new sample to a time series.
- Args:
- key:
- time-series key
- timestamp:
- Timestamp of the sample. * can be used for automatic timestamp (using the system clock).
- value:
- Numeric data value of the sample
- retention_msecs:
- Maximum retention period, compared to maximal existing timestamp (in milliseconds).
- If None or 0 is passed then the series is not trimmed at all.
- uncompressed:
- Changes data storage from compressed (by default) to uncompressed
- labels:
- Set of label-value pairs that represent metadata labels of the key.
- chunk_size:
- Memory size, in bytes, allocated for each data chunk.
- Must be a multiple of 8 in the range [128 .. 1048576].
- duplicate_policy:
- Policy for handling multiple samples with identical timestamps.
- Can be one of:
- - 'block': an error will occur for any out of order sample.
- - 'first': ignore the new value.
- - 'last': override with latest value.
- - 'min': only override if the value is lower than the existing value.
- - 'max': only override if the value is higher than the existing value.
- For more information: https://redis.io/commands/ts.add/
- """ # noqa
- params = [key, timestamp, value]
- self._append_retention(params, retention_msecs)
- self._append_uncompressed(params, uncompressed)
- self._append_chunk_size(params, chunk_size)
- self._append_duplicate_policy(params, ADD_CMD, duplicate_policy)
- self._append_labels(params, labels)
- return self.execute_command(ADD_CMD, *params)
- def madd(self, ktv_tuples: List[Tuple[KeyT, Union[int, str], Number]]):
- """
- Append (or create and append) a new `value` to series
- `key` with `timestamp`.
- Expects a list of `tuples` as (`key`,`timestamp`, `value`).
- Return value is an array with timestamps of insertions.
- For more information: https://redis.io/commands/ts.madd/
- """ # noqa
- params = []
- for ktv in ktv_tuples:
- params.extend(ktv)
- return self.execute_command(MADD_CMD, *params)
- def incrby(
- self,
- key: KeyT,
- value: Number,
- timestamp: Optional[Union[int, str]] = None,
- retention_msecs: Optional[int] = None,
- uncompressed: Optional[bool] = False,
- labels: Optional[Dict[str, str]] = None,
- chunk_size: Optional[int] = None,
- ):
- """
- Increment (or create an time-series and increment) the latest sample's of a series.
- This command can be used as a counter or gauge that automatically gets history as a time series.
- Args:
- key:
- time-series key
- value:
- Numeric data value of the sample
- timestamp:
- Timestamp of the sample. * can be used for automatic timestamp (using the system clock).
- retention_msecs:
- Maximum age for samples compared to last event time (in milliseconds).
- If None or 0 is passed then the series is not trimmed at all.
- uncompressed:
- Changes data storage from compressed (by default) to uncompressed
- labels:
- Set of label-value pairs that represent metadata labels of the key.
- chunk_size:
- Memory size, in bytes, allocated for each data chunk.
- For more information: https://redis.io/commands/ts.incrby/
- """ # noqa
- params = [key, value]
- self._append_timestamp(params, timestamp)
- self._append_retention(params, retention_msecs)
- self._append_uncompressed(params, uncompressed)
- self._append_chunk_size(params, chunk_size)
- self._append_labels(params, labels)
- return self.execute_command(INCRBY_CMD, *params)
- def decrby(
- self,
- key: KeyT,
- value: Number,
- timestamp: Optional[Union[int, str]] = None,
- retention_msecs: Optional[int] = None,
- uncompressed: Optional[bool] = False,
- labels: Optional[Dict[str, str]] = None,
- chunk_size: Optional[int] = None,
- ):
- """
- Decrement (or create an time-series and decrement) the latest sample's of a series.
- This command can be used as a counter or gauge that automatically gets history as a time series.
- Args:
- key:
- time-series key
- value:
- Numeric data value of the sample
- timestamp:
- Timestamp of the sample. * can be used for automatic timestamp (using the system clock).
- retention_msecs:
- Maximum age for samples compared to last event time (in milliseconds).
- If None or 0 is passed then the series is not trimmed at all.
- uncompressed:
- Changes data storage from compressed (by default) to uncompressed
- labels:
- Set of label-value pairs that represent metadata labels of the key.
- chunk_size:
- Memory size, in bytes, allocated for each data chunk.
- For more information: https://redis.io/commands/ts.decrby/
- """ # noqa
- params = [key, value]
- self._append_timestamp(params, timestamp)
- self._append_retention(params, retention_msecs)
- self._append_uncompressed(params, uncompressed)
- self._append_chunk_size(params, chunk_size)
- self._append_labels(params, labels)
- return self.execute_command(DECRBY_CMD, *params)
- def delete(self, key: KeyT, from_time: int, to_time: int):
- """
- Delete all samples between two timestamps for a given time series.
- Args:
- key:
- time-series key.
- from_time:
- Start timestamp for the range deletion.
- to_time:
- End timestamp for the range deletion.
- For more information: https://redis.io/commands/ts.del/
- """ # noqa
- return self.execute_command(DEL_CMD, key, from_time, to_time)
- def createrule(
- self,
- source_key: KeyT,
- dest_key: KeyT,
- aggregation_type: str,
- bucket_size_msec: int,
- align_timestamp: Optional[int] = None,
- ):
- """
- Create a compaction rule from values added to `source_key` into `dest_key`.
- Args:
- source_key:
- Key name for source time series
- dest_key:
- Key name for destination (compacted) time series
- aggregation_type:
- Aggregation type: One of the following:
- [`avg`, `sum`, `min`, `max`, `range`, `count`, `first`, `last`, `std.p`,
- `std.s`, `var.p`, `var.s`, `twa`]
- bucket_size_msec:
- Duration of each bucket, in milliseconds
- align_timestamp:
- Assure that there is a bucket that starts at exactly align_timestamp and
- align all other buckets accordingly.
- For more information: https://redis.io/commands/ts.createrule/
- """ # noqa
- params = [source_key, dest_key]
- self._append_aggregation(params, aggregation_type, bucket_size_msec)
- if align_timestamp is not None:
- params.append(align_timestamp)
- return self.execute_command(CREATERULE_CMD, *params)
- def deleterule(self, source_key: KeyT, dest_key: KeyT):
- """
- Delete a compaction rule from `source_key` to `dest_key`..
- For more information: https://redis.io/commands/ts.deleterule/
- """ # noqa
- return self.execute_command(DELETERULE_CMD, source_key, dest_key)
- def __range_params(
- self,
- key: KeyT,
- from_time: Union[int, str],
- to_time: Union[int, str],
- count: Optional[int],
- aggregation_type: Optional[str],
- bucket_size_msec: Optional[int],
- filter_by_ts: Optional[List[int]],
- filter_by_min_value: Optional[int],
- filter_by_max_value: Optional[int],
- align: Optional[Union[int, str]],
- latest: Optional[bool],
- bucket_timestamp: Optional[str],
- empty: Optional[bool],
- ):
- """Create TS.RANGE and TS.REVRANGE arguments."""
- params = [key, from_time, to_time]
- self._append_latest(params, latest)
- self._append_filer_by_ts(params, filter_by_ts)
- self._append_filer_by_value(params, filter_by_min_value, filter_by_max_value)
- self._append_count(params, count)
- self._append_align(params, align)
- self._append_aggregation(params, aggregation_type, bucket_size_msec)
- self._append_bucket_timestamp(params, bucket_timestamp)
- self._append_empty(params, empty)
- return params
- def range(
- self,
- key: KeyT,
- from_time: Union[int, str],
- to_time: Union[int, str],
- count: Optional[int] = None,
- aggregation_type: Optional[str] = None,
- bucket_size_msec: Optional[int] = 0,
- filter_by_ts: Optional[List[int]] = None,
- filter_by_min_value: Optional[int] = None,
- filter_by_max_value: Optional[int] = None,
- align: Optional[Union[int, str]] = None,
- latest: Optional[bool] = False,
- bucket_timestamp: Optional[str] = None,
- empty: Optional[bool] = False,
- ):
- """
- Query a range in forward direction for a specific time-serie.
- Args:
- key:
- Key name for timeseries.
- from_time:
- Start timestamp for the range query. - can be used to express the minimum possible timestamp (0).
- to_time:
- End timestamp for range query, + can be used to express the maximum possible timestamp.
- count:
- Limits the number of returned samples.
- aggregation_type:
- Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
- `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
- bucket_size_msec:
- Time bucket for aggregation in milliseconds.
- filter_by_ts:
- List of timestamps to filter the result by specific timestamps.
- filter_by_min_value:
- Filter result by minimum value (must mention also filter by_max_value).
- filter_by_max_value:
- Filter result by maximum value (must mention also filter by_min_value).
- align:
- Timestamp for alignment control for aggregation.
- latest:
- Used when a time series is a compaction, reports the compacted value of the
- latest possibly partial bucket
- bucket_timestamp:
- Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
- `high`, `~`, `mid`].
- empty:
- Reports aggregations for empty buckets.
- For more information: https://redis.io/commands/ts.range/
- """ # noqa
- params = self.__range_params(
- key,
- from_time,
- to_time,
- count,
- aggregation_type,
- bucket_size_msec,
- filter_by_ts,
- filter_by_min_value,
- filter_by_max_value,
- align,
- latest,
- bucket_timestamp,
- empty,
- )
- return self.execute_command(RANGE_CMD, *params)
- def revrange(
- self,
- key: KeyT,
- from_time: Union[int, str],
- to_time: Union[int, str],
- count: Optional[int] = None,
- aggregation_type: Optional[str] = None,
- bucket_size_msec: Optional[int] = 0,
- filter_by_ts: Optional[List[int]] = None,
- filter_by_min_value: Optional[int] = None,
- filter_by_max_value: Optional[int] = None,
- align: Optional[Union[int, str]] = None,
- latest: Optional[bool] = False,
- bucket_timestamp: Optional[str] = None,
- empty: Optional[bool] = False,
- ):
- """
- Query a range in reverse direction for a specific time-series.
- **Note**: This command is only available since RedisTimeSeries >= v1.4
- Args:
- key:
- Key name for timeseries.
- from_time:
- Start timestamp for the range query. - can be used to express the minimum possible timestamp (0).
- to_time:
- End timestamp for range query, + can be used to express the maximum possible timestamp.
- count:
- Limits the number of returned samples.
- aggregation_type:
- Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
- `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
- bucket_size_msec:
- Time bucket for aggregation in milliseconds.
- filter_by_ts:
- List of timestamps to filter the result by specific timestamps.
- filter_by_min_value:
- Filter result by minimum value (must mention also filter_by_max_value).
- filter_by_max_value:
- Filter result by maximum value (must mention also filter_by_min_value).
- align:
- Timestamp for alignment control for aggregation.
- latest:
- Used when a time series is a compaction, reports the compacted value of the
- latest possibly partial bucket
- bucket_timestamp:
- Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
- `high`, `~`, `mid`].
- empty:
- Reports aggregations for empty buckets.
- For more information: https://redis.io/commands/ts.revrange/
- """ # noqa
- params = self.__range_params(
- key,
- from_time,
- to_time,
- count,
- aggregation_type,
- bucket_size_msec,
- filter_by_ts,
- filter_by_min_value,
- filter_by_max_value,
- align,
- latest,
- bucket_timestamp,
- empty,
- )
- return self.execute_command(REVRANGE_CMD, *params)
- def __mrange_params(
- self,
- aggregation_type: Optional[str],
- bucket_size_msec: Optional[int],
- count: Optional[int],
- filters: List[str],
- from_time: Union[int, str],
- to_time: Union[int, str],
- with_labels: Optional[bool],
- filter_by_ts: Optional[List[int]],
- filter_by_min_value: Optional[int],
- filter_by_max_value: Optional[int],
- groupby: Optional[str],
- reduce: Optional[str],
- select_labels: Optional[List[str]],
- align: Optional[Union[int, str]],
- latest: Optional[bool],
- bucket_timestamp: Optional[str],
- empty: Optional[bool],
- ):
- """Create TS.MRANGE and TS.MREVRANGE arguments."""
- params = [from_time, to_time]
- self._append_latest(params, latest)
- self._append_filer_by_ts(params, filter_by_ts)
- self._append_filer_by_value(params, filter_by_min_value, filter_by_max_value)
- self._append_with_labels(params, with_labels, select_labels)
- self._append_count(params, count)
- self._append_align(params, align)
- self._append_aggregation(params, aggregation_type, bucket_size_msec)
- self._append_bucket_timestamp(params, bucket_timestamp)
- self._append_empty(params, empty)
- params.extend(["FILTER"])
- params += filters
- self._append_groupby_reduce(params, groupby, reduce)
- return params
- def mrange(
- self,
- from_time: Union[int, str],
- to_time: Union[int, str],
- filters: List[str],
- count: Optional[int] = None,
- aggregation_type: Optional[str] = None,
- bucket_size_msec: Optional[int] = 0,
- with_labels: Optional[bool] = False,
- filter_by_ts: Optional[List[int]] = None,
- filter_by_min_value: Optional[int] = None,
- filter_by_max_value: Optional[int] = None,
- groupby: Optional[str] = None,
- reduce: Optional[str] = None,
- select_labels: Optional[List[str]] = None,
- align: Optional[Union[int, str]] = None,
- latest: Optional[bool] = False,
- bucket_timestamp: Optional[str] = None,
- empty: Optional[bool] = False,
- ):
- """
- Query a range across multiple time-series by filters in forward direction.
- Args:
- from_time:
- Start timestamp for the range query. `-` can be used to express the minimum possible timestamp (0).
- to_time:
- End timestamp for range query, `+` can be used to express the maximum possible timestamp.
- filters:
- filter to match the time-series labels.
- count:
- Limits the number of returned samples.
- aggregation_type:
- Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
- `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
- bucket_size_msec:
- Time bucket for aggregation in milliseconds.
- with_labels:
- Include in the reply all label-value pairs representing metadata labels of the time series.
- filter_by_ts:
- List of timestamps to filter the result by specific timestamps.
- filter_by_min_value:
- Filter result by minimum value (must mention also filter_by_max_value).
- filter_by_max_value:
- Filter result by maximum value (must mention also filter_by_min_value).
- groupby:
- Grouping by fields the results (must mention also reduce).
- reduce:
- Applying reducer functions on each group. Can be one of [`avg` `sum`, `min`,
- `max`, `range`, `count`, `std.p`, `std.s`, `var.p`, `var.s`].
- select_labels:
- Include in the reply only a subset of the key-value pair labels of a series.
- align:
- Timestamp for alignment control for aggregation.
- latest:
- Used when a time series is a compaction, reports the compacted
- value of the latest possibly partial bucket
- bucket_timestamp:
- Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
- `high`, `~`, `mid`].
- empty:
- Reports aggregations for empty buckets.
- For more information: https://redis.io/commands/ts.mrange/
- """ # noqa
- params = self.__mrange_params(
- aggregation_type,
- bucket_size_msec,
- count,
- filters,
- from_time,
- to_time,
- with_labels,
- filter_by_ts,
- filter_by_min_value,
- filter_by_max_value,
- groupby,
- reduce,
- select_labels,
- align,
- latest,
- bucket_timestamp,
- empty,
- )
- return self.execute_command(MRANGE_CMD, *params)
- def mrevrange(
- self,
- from_time: Union[int, str],
- to_time: Union[int, str],
- filters: List[str],
- count: Optional[int] = None,
- aggregation_type: Optional[str] = None,
- bucket_size_msec: Optional[int] = 0,
- with_labels: Optional[bool] = False,
- filter_by_ts: Optional[List[int]] = None,
- filter_by_min_value: Optional[int] = None,
- filter_by_max_value: Optional[int] = None,
- groupby: Optional[str] = None,
- reduce: Optional[str] = None,
- select_labels: Optional[List[str]] = None,
- align: Optional[Union[int, str]] = None,
- latest: Optional[bool] = False,
- bucket_timestamp: Optional[str] = None,
- empty: Optional[bool] = False,
- ):
- """
- Query a range across multiple time-series by filters in reverse direction.
- Args:
- from_time:
- Start timestamp for the range query. - can be used to express the minimum possible timestamp (0).
- to_time:
- End timestamp for range query, + can be used to express the maximum possible timestamp.
- filters:
- Filter to match the time-series labels.
- count:
- Limits the number of returned samples.
- aggregation_type:
- Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
- `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
- bucket_size_msec:
- Time bucket for aggregation in milliseconds.
- with_labels:
- Include in the reply all label-value pairs representing metadata labels of the time series.
- filter_by_ts:
- List of timestamps to filter the result by specific timestamps.
- filter_by_min_value:
- Filter result by minimum value (must mention also filter_by_max_value).
- filter_by_max_value:
- Filter result by maximum value (must mention also filter_by_min_value).
- groupby:
- Grouping by fields the results (must mention also reduce).
- reduce:
- Applying reducer functions on each group. Can be one of [`avg` `sum`, `min`,
- `max`, `range`, `count`, `std.p`, `std.s`, `var.p`, `var.s`].
- select_labels:
- Include in the reply only a subset of the key-value pair labels of a series.
- align:
- Timestamp for alignment control for aggregation.
- latest:
- Used when a time series is a compaction, reports the compacted
- value of the latest possibly partial bucket
- bucket_timestamp:
- Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
- `high`, `~`, `mid`].
- empty:
- Reports aggregations for empty buckets.
- For more information: https://redis.io/commands/ts.mrevrange/
- """ # noqa
- params = self.__mrange_params(
- aggregation_type,
- bucket_size_msec,
- count,
- filters,
- from_time,
- to_time,
- with_labels,
- filter_by_ts,
- filter_by_min_value,
- filter_by_max_value,
- groupby,
- reduce,
- select_labels,
- align,
- latest,
- bucket_timestamp,
- empty,
- )
- return self.execute_command(MREVRANGE_CMD, *params)
- def get(self, key: KeyT, latest: Optional[bool] = False):
- """# noqa
- Get the last sample of `key`.
- `latest` used when a time series is a compaction, reports the compacted
- value of the latest (possibly partial) bucket
- For more information: https://redis.io/commands/ts.get/
- """ # noqa
- params = [key]
- self._append_latest(params, latest)
- return self.execute_command(GET_CMD, *params)
- def mget(
- self,
- filters: List[str],
- with_labels: Optional[bool] = False,
- select_labels: Optional[List[str]] = None,
- latest: Optional[bool] = False,
- ):
- """# noqa
- Get the last samples matching the specific `filter`.
- Args:
- filters:
- Filter to match the time-series labels.
- with_labels:
- Include in the reply all label-value pairs representing metadata
- labels of the time series.
- select_labels:
- Include in the reply only a subset of the key-value pair labels of a series.
- latest:
- Used when a time series is a compaction, reports the compacted
- value of the latest possibly partial bucket
- For more information: https://redis.io/commands/ts.mget/
- """ # noqa
- params = []
- self._append_latest(params, latest)
- self._append_with_labels(params, with_labels, select_labels)
- params.extend(["FILTER"])
- params += filters
- return self.execute_command(MGET_CMD, *params)
- def info(self, key: KeyT):
- """# noqa
- Get information of `key`.
- For more information: https://redis.io/commands/ts.info/
- """ # noqa
- return self.execute_command(INFO_CMD, key)
- def queryindex(self, filters: List[str]):
- """# noqa
- Get all time series keys matching the `filter` list.
- For more information: https://redis.io/commands/ts.queryindex/
- """ # noq
- return self.execute_command(QUERYINDEX_CMD, *filters)
- @staticmethod
- def _append_uncompressed(params: List[str], uncompressed: Optional[bool]):
- """Append UNCOMPRESSED tag to params."""
- if uncompressed:
- params.extend(["UNCOMPRESSED"])
- @staticmethod
- def _append_with_labels(
- params: List[str],
- with_labels: Optional[bool],
- select_labels: Optional[List[str]],
- ):
- """Append labels behavior to params."""
- if with_labels and select_labels:
- raise DataError(
- "with_labels and select_labels cannot be provided together."
- )
- if with_labels:
- params.extend(["WITHLABELS"])
- if select_labels:
- params.extend(["SELECTED_LABELS", *select_labels])
- @staticmethod
- def _append_groupby_reduce(
- params: List[str], groupby: Optional[str], reduce: Optional[str]
- ):
- """Append GROUPBY REDUCE property to params."""
- if groupby is not None and reduce is not None:
- params.extend(["GROUPBY", groupby, "REDUCE", reduce.upper()])
- @staticmethod
- def _append_retention(params: List[str], retention: Optional[int]):
- """Append RETENTION property to params."""
- if retention is not None:
- params.extend(["RETENTION", retention])
- @staticmethod
- def _append_labels(params: List[str], labels: Optional[List[str]]):
- """Append LABELS property to params."""
- if labels:
- params.append("LABELS")
- for k, v in labels.items():
- params.extend([k, v])
- @staticmethod
- def _append_count(params: List[str], count: Optional[int]):
- """Append COUNT property to params."""
- if count is not None:
- params.extend(["COUNT", count])
- @staticmethod
- def _append_timestamp(params: List[str], timestamp: Optional[int]):
- """Append TIMESTAMP property to params."""
- if timestamp is not None:
- params.extend(["TIMESTAMP", timestamp])
- @staticmethod
- def _append_align(params: List[str], align: Optional[Union[int, str]]):
- """Append ALIGN property to params."""
- if align is not None:
- params.extend(["ALIGN", align])
- @staticmethod
- def _append_aggregation(
- params: List[str],
- aggregation_type: Optional[str],
- bucket_size_msec: Optional[int],
- ):
- """Append AGGREGATION property to params."""
- if aggregation_type is not None:
- params.extend(["AGGREGATION", aggregation_type, bucket_size_msec])
- @staticmethod
- def _append_chunk_size(params: List[str], chunk_size: Optional[int]):
- """Append CHUNK_SIZE property to params."""
- if chunk_size is not None:
- params.extend(["CHUNK_SIZE", chunk_size])
- @staticmethod
- def _append_duplicate_policy(
- params: List[str], command: Optional[str], duplicate_policy: Optional[str]
- ):
- """Append DUPLICATE_POLICY property to params on CREATE
- and ON_DUPLICATE on ADD.
- """
- if duplicate_policy is not None:
- if command == "TS.ADD":
- params.extend(["ON_DUPLICATE", duplicate_policy])
- else:
- params.extend(["DUPLICATE_POLICY", duplicate_policy])
- @staticmethod
- def _append_filer_by_ts(params: List[str], ts_list: Optional[List[int]]):
- """Append FILTER_BY_TS property to params."""
- if ts_list is not None:
- params.extend(["FILTER_BY_TS", *ts_list])
- @staticmethod
- def _append_filer_by_value(
- params: List[str], min_value: Optional[int], max_value: Optional[int]
- ):
- """Append FILTER_BY_VALUE property to params."""
- if min_value is not None and max_value is not None:
- params.extend(["FILTER_BY_VALUE", min_value, max_value])
- @staticmethod
- def _append_latest(params: List[str], latest: Optional[bool]):
- """Append LATEST property to params."""
- if latest:
- params.append("LATEST")
- @staticmethod
- def _append_bucket_timestamp(params: List[str], bucket_timestamp: Optional[str]):
- """Append BUCKET_TIMESTAMP property to params."""
- if bucket_timestamp is not None:
- params.extend(["BUCKETTIMESTAMP", bucket_timestamp])
- @staticmethod
- def _append_empty(params: List[str], empty: Optional[bool]):
- """Append EMPTY property to params."""
- if empty:
- params.append("EMPTY")
|