commands.py 33 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896
  1. from typing import Dict, List, Optional, Tuple, Union
  2. from redis.exceptions import DataError
  3. from redis.typing import KeyT, Number
  4. ADD_CMD = "TS.ADD"
  5. ALTER_CMD = "TS.ALTER"
  6. CREATERULE_CMD = "TS.CREATERULE"
  7. CREATE_CMD = "TS.CREATE"
  8. DECRBY_CMD = "TS.DECRBY"
  9. DELETERULE_CMD = "TS.DELETERULE"
  10. DEL_CMD = "TS.DEL"
  11. GET_CMD = "TS.GET"
  12. INCRBY_CMD = "TS.INCRBY"
  13. INFO_CMD = "TS.INFO"
  14. MADD_CMD = "TS.MADD"
  15. MGET_CMD = "TS.MGET"
  16. MRANGE_CMD = "TS.MRANGE"
  17. MREVRANGE_CMD = "TS.MREVRANGE"
  18. QUERYINDEX_CMD = "TS.QUERYINDEX"
  19. RANGE_CMD = "TS.RANGE"
  20. REVRANGE_CMD = "TS.REVRANGE"
  21. class TimeSeriesCommands:
  22. """RedisTimeSeries Commands."""
  23. def create(
  24. self,
  25. key: KeyT,
  26. retention_msecs: Optional[int] = None,
  27. uncompressed: Optional[bool] = False,
  28. labels: Optional[Dict[str, str]] = None,
  29. chunk_size: Optional[int] = None,
  30. duplicate_policy: Optional[str] = None,
  31. ):
  32. """
  33. Create a new time-series.
  34. Args:
  35. key:
  36. time-series key
  37. retention_msecs:
  38. Maximum age for samples compared to highest reported timestamp (in milliseconds).
  39. If None or 0 is passed then the series is not trimmed at all.
  40. uncompressed:
  41. Changes data storage from compressed (by default) to uncompressed
  42. labels:
  43. Set of label-value pairs that represent metadata labels of the key.
  44. chunk_size:
  45. Memory size, in bytes, allocated for each data chunk.
  46. Must be a multiple of 8 in the range [128 .. 1048576].
  47. duplicate_policy:
  48. Policy for handling multiple samples with identical timestamps.
  49. Can be one of:
  50. - 'block': an error will occur for any out of order sample.
  51. - 'first': ignore the new value.
  52. - 'last': override with latest value.
  53. - 'min': only override if the value is lower than the existing value.
  54. - 'max': only override if the value is higher than the existing value.
  55. For more information: https://redis.io/commands/ts.create/
  56. """ # noqa
  57. params = [key]
  58. self._append_retention(params, retention_msecs)
  59. self._append_uncompressed(params, uncompressed)
  60. self._append_chunk_size(params, chunk_size)
  61. self._append_duplicate_policy(params, CREATE_CMD, duplicate_policy)
  62. self._append_labels(params, labels)
  63. return self.execute_command(CREATE_CMD, *params)
  64. def alter(
  65. self,
  66. key: KeyT,
  67. retention_msecs: Optional[int] = None,
  68. labels: Optional[Dict[str, str]] = None,
  69. chunk_size: Optional[int] = None,
  70. duplicate_policy: Optional[str] = None,
  71. ):
  72. """
  73. Update the retention, chunk size, duplicate policy, and labels of an existing
  74. time series.
  75. Args:
  76. key:
  77. time-series key
  78. retention_msecs:
  79. Maximum retention period, compared to maximal existing timestamp (in milliseconds).
  80. If None or 0 is passed then the series is not trimmed at all.
  81. labels:
  82. Set of label-value pairs that represent metadata labels of the key.
  83. chunk_size:
  84. Memory size, in bytes, allocated for each data chunk.
  85. Must be a multiple of 8 in the range [128 .. 1048576].
  86. duplicate_policy:
  87. Policy for handling multiple samples with identical timestamps.
  88. Can be one of:
  89. - 'block': an error will occur for any out of order sample.
  90. - 'first': ignore the new value.
  91. - 'last': override with latest value.
  92. - 'min': only override if the value is lower than the existing value.
  93. - 'max': only override if the value is higher than the existing value.
  94. For more information: https://redis.io/commands/ts.alter/
  95. """ # noqa
  96. params = [key]
  97. self._append_retention(params, retention_msecs)
  98. self._append_chunk_size(params, chunk_size)
  99. self._append_duplicate_policy(params, ALTER_CMD, duplicate_policy)
  100. self._append_labels(params, labels)
  101. return self.execute_command(ALTER_CMD, *params)
  102. def add(
  103. self,
  104. key: KeyT,
  105. timestamp: Union[int, str],
  106. value: Number,
  107. retention_msecs: Optional[int] = None,
  108. uncompressed: Optional[bool] = False,
  109. labels: Optional[Dict[str, str]] = None,
  110. chunk_size: Optional[int] = None,
  111. duplicate_policy: Optional[str] = None,
  112. ):
  113. """
  114. Append (or create and append) a new sample to a time series.
  115. Args:
  116. key:
  117. time-series key
  118. timestamp:
  119. Timestamp of the sample. * can be used for automatic timestamp (using the system clock).
  120. value:
  121. Numeric data value of the sample
  122. retention_msecs:
  123. Maximum retention period, compared to maximal existing timestamp (in milliseconds).
  124. If None or 0 is passed then the series is not trimmed at all.
  125. uncompressed:
  126. Changes data storage from compressed (by default) to uncompressed
  127. labels:
  128. Set of label-value pairs that represent metadata labels of the key.
  129. chunk_size:
  130. Memory size, in bytes, allocated for each data chunk.
  131. Must be a multiple of 8 in the range [128 .. 1048576].
  132. duplicate_policy:
  133. Policy for handling multiple samples with identical timestamps.
  134. Can be one of:
  135. - 'block': an error will occur for any out of order sample.
  136. - 'first': ignore the new value.
  137. - 'last': override with latest value.
  138. - 'min': only override if the value is lower than the existing value.
  139. - 'max': only override if the value is higher than the existing value.
  140. For more information: https://redis.io/commands/ts.add/
  141. """ # noqa
  142. params = [key, timestamp, value]
  143. self._append_retention(params, retention_msecs)
  144. self._append_uncompressed(params, uncompressed)
  145. self._append_chunk_size(params, chunk_size)
  146. self._append_duplicate_policy(params, ADD_CMD, duplicate_policy)
  147. self._append_labels(params, labels)
  148. return self.execute_command(ADD_CMD, *params)
  149. def madd(self, ktv_tuples: List[Tuple[KeyT, Union[int, str], Number]]):
  150. """
  151. Append (or create and append) a new `value` to series
  152. `key` with `timestamp`.
  153. Expects a list of `tuples` as (`key`,`timestamp`, `value`).
  154. Return value is an array with timestamps of insertions.
  155. For more information: https://redis.io/commands/ts.madd/
  156. """ # noqa
  157. params = []
  158. for ktv in ktv_tuples:
  159. params.extend(ktv)
  160. return self.execute_command(MADD_CMD, *params)
  161. def incrby(
  162. self,
  163. key: KeyT,
  164. value: Number,
  165. timestamp: Optional[Union[int, str]] = None,
  166. retention_msecs: Optional[int] = None,
  167. uncompressed: Optional[bool] = False,
  168. labels: Optional[Dict[str, str]] = None,
  169. chunk_size: Optional[int] = None,
  170. ):
  171. """
  172. Increment (or create an time-series and increment) the latest sample's of a series.
  173. This command can be used as a counter or gauge that automatically gets history as a time series.
  174. Args:
  175. key:
  176. time-series key
  177. value:
  178. Numeric data value of the sample
  179. timestamp:
  180. Timestamp of the sample. * can be used for automatic timestamp (using the system clock).
  181. retention_msecs:
  182. Maximum age for samples compared to last event time (in milliseconds).
  183. If None or 0 is passed then the series is not trimmed at all.
  184. uncompressed:
  185. Changes data storage from compressed (by default) to uncompressed
  186. labels:
  187. Set of label-value pairs that represent metadata labels of the key.
  188. chunk_size:
  189. Memory size, in bytes, allocated for each data chunk.
  190. For more information: https://redis.io/commands/ts.incrby/
  191. """ # noqa
  192. params = [key, value]
  193. self._append_timestamp(params, timestamp)
  194. self._append_retention(params, retention_msecs)
  195. self._append_uncompressed(params, uncompressed)
  196. self._append_chunk_size(params, chunk_size)
  197. self._append_labels(params, labels)
  198. return self.execute_command(INCRBY_CMD, *params)
  199. def decrby(
  200. self,
  201. key: KeyT,
  202. value: Number,
  203. timestamp: Optional[Union[int, str]] = None,
  204. retention_msecs: Optional[int] = None,
  205. uncompressed: Optional[bool] = False,
  206. labels: Optional[Dict[str, str]] = None,
  207. chunk_size: Optional[int] = None,
  208. ):
  209. """
  210. Decrement (or create an time-series and decrement) the latest sample's of a series.
  211. This command can be used as a counter or gauge that automatically gets history as a time series.
  212. Args:
  213. key:
  214. time-series key
  215. value:
  216. Numeric data value of the sample
  217. timestamp:
  218. Timestamp of the sample. * can be used for automatic timestamp (using the system clock).
  219. retention_msecs:
  220. Maximum age for samples compared to last event time (in milliseconds).
  221. If None or 0 is passed then the series is not trimmed at all.
  222. uncompressed:
  223. Changes data storage from compressed (by default) to uncompressed
  224. labels:
  225. Set of label-value pairs that represent metadata labels of the key.
  226. chunk_size:
  227. Memory size, in bytes, allocated for each data chunk.
  228. For more information: https://redis.io/commands/ts.decrby/
  229. """ # noqa
  230. params = [key, value]
  231. self._append_timestamp(params, timestamp)
  232. self._append_retention(params, retention_msecs)
  233. self._append_uncompressed(params, uncompressed)
  234. self._append_chunk_size(params, chunk_size)
  235. self._append_labels(params, labels)
  236. return self.execute_command(DECRBY_CMD, *params)
  237. def delete(self, key: KeyT, from_time: int, to_time: int):
  238. """
  239. Delete all samples between two timestamps for a given time series.
  240. Args:
  241. key:
  242. time-series key.
  243. from_time:
  244. Start timestamp for the range deletion.
  245. to_time:
  246. End timestamp for the range deletion.
  247. For more information: https://redis.io/commands/ts.del/
  248. """ # noqa
  249. return self.execute_command(DEL_CMD, key, from_time, to_time)
  250. def createrule(
  251. self,
  252. source_key: KeyT,
  253. dest_key: KeyT,
  254. aggregation_type: str,
  255. bucket_size_msec: int,
  256. align_timestamp: Optional[int] = None,
  257. ):
  258. """
  259. Create a compaction rule from values added to `source_key` into `dest_key`.
  260. Args:
  261. source_key:
  262. Key name for source time series
  263. dest_key:
  264. Key name for destination (compacted) time series
  265. aggregation_type:
  266. Aggregation type: One of the following:
  267. [`avg`, `sum`, `min`, `max`, `range`, `count`, `first`, `last`, `std.p`,
  268. `std.s`, `var.p`, `var.s`, `twa`]
  269. bucket_size_msec:
  270. Duration of each bucket, in milliseconds
  271. align_timestamp:
  272. Assure that there is a bucket that starts at exactly align_timestamp and
  273. align all other buckets accordingly.
  274. For more information: https://redis.io/commands/ts.createrule/
  275. """ # noqa
  276. params = [source_key, dest_key]
  277. self._append_aggregation(params, aggregation_type, bucket_size_msec)
  278. if align_timestamp is not None:
  279. params.append(align_timestamp)
  280. return self.execute_command(CREATERULE_CMD, *params)
  281. def deleterule(self, source_key: KeyT, dest_key: KeyT):
  282. """
  283. Delete a compaction rule from `source_key` to `dest_key`..
  284. For more information: https://redis.io/commands/ts.deleterule/
  285. """ # noqa
  286. return self.execute_command(DELETERULE_CMD, source_key, dest_key)
  287. def __range_params(
  288. self,
  289. key: KeyT,
  290. from_time: Union[int, str],
  291. to_time: Union[int, str],
  292. count: Optional[int],
  293. aggregation_type: Optional[str],
  294. bucket_size_msec: Optional[int],
  295. filter_by_ts: Optional[List[int]],
  296. filter_by_min_value: Optional[int],
  297. filter_by_max_value: Optional[int],
  298. align: Optional[Union[int, str]],
  299. latest: Optional[bool],
  300. bucket_timestamp: Optional[str],
  301. empty: Optional[bool],
  302. ):
  303. """Create TS.RANGE and TS.REVRANGE arguments."""
  304. params = [key, from_time, to_time]
  305. self._append_latest(params, latest)
  306. self._append_filer_by_ts(params, filter_by_ts)
  307. self._append_filer_by_value(params, filter_by_min_value, filter_by_max_value)
  308. self._append_count(params, count)
  309. self._append_align(params, align)
  310. self._append_aggregation(params, aggregation_type, bucket_size_msec)
  311. self._append_bucket_timestamp(params, bucket_timestamp)
  312. self._append_empty(params, empty)
  313. return params
  314. def range(
  315. self,
  316. key: KeyT,
  317. from_time: Union[int, str],
  318. to_time: Union[int, str],
  319. count: Optional[int] = None,
  320. aggregation_type: Optional[str] = None,
  321. bucket_size_msec: Optional[int] = 0,
  322. filter_by_ts: Optional[List[int]] = None,
  323. filter_by_min_value: Optional[int] = None,
  324. filter_by_max_value: Optional[int] = None,
  325. align: Optional[Union[int, str]] = None,
  326. latest: Optional[bool] = False,
  327. bucket_timestamp: Optional[str] = None,
  328. empty: Optional[bool] = False,
  329. ):
  330. """
  331. Query a range in forward direction for a specific time-serie.
  332. Args:
  333. key:
  334. Key name for timeseries.
  335. from_time:
  336. Start timestamp for the range query. - can be used to express the minimum possible timestamp (0).
  337. to_time:
  338. End timestamp for range query, + can be used to express the maximum possible timestamp.
  339. count:
  340. Limits the number of returned samples.
  341. aggregation_type:
  342. Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
  343. `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
  344. bucket_size_msec:
  345. Time bucket for aggregation in milliseconds.
  346. filter_by_ts:
  347. List of timestamps to filter the result by specific timestamps.
  348. filter_by_min_value:
  349. Filter result by minimum value (must mention also filter by_max_value).
  350. filter_by_max_value:
  351. Filter result by maximum value (must mention also filter by_min_value).
  352. align:
  353. Timestamp for alignment control for aggregation.
  354. latest:
  355. Used when a time series is a compaction, reports the compacted value of the
  356. latest possibly partial bucket
  357. bucket_timestamp:
  358. Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
  359. `high`, `~`, `mid`].
  360. empty:
  361. Reports aggregations for empty buckets.
  362. For more information: https://redis.io/commands/ts.range/
  363. """ # noqa
  364. params = self.__range_params(
  365. key,
  366. from_time,
  367. to_time,
  368. count,
  369. aggregation_type,
  370. bucket_size_msec,
  371. filter_by_ts,
  372. filter_by_min_value,
  373. filter_by_max_value,
  374. align,
  375. latest,
  376. bucket_timestamp,
  377. empty,
  378. )
  379. return self.execute_command(RANGE_CMD, *params)
  380. def revrange(
  381. self,
  382. key: KeyT,
  383. from_time: Union[int, str],
  384. to_time: Union[int, str],
  385. count: Optional[int] = None,
  386. aggregation_type: Optional[str] = None,
  387. bucket_size_msec: Optional[int] = 0,
  388. filter_by_ts: Optional[List[int]] = None,
  389. filter_by_min_value: Optional[int] = None,
  390. filter_by_max_value: Optional[int] = None,
  391. align: Optional[Union[int, str]] = None,
  392. latest: Optional[bool] = False,
  393. bucket_timestamp: Optional[str] = None,
  394. empty: Optional[bool] = False,
  395. ):
  396. """
  397. Query a range in reverse direction for a specific time-series.
  398. **Note**: This command is only available since RedisTimeSeries >= v1.4
  399. Args:
  400. key:
  401. Key name for timeseries.
  402. from_time:
  403. Start timestamp for the range query. - can be used to express the minimum possible timestamp (0).
  404. to_time:
  405. End timestamp for range query, + can be used to express the maximum possible timestamp.
  406. count:
  407. Limits the number of returned samples.
  408. aggregation_type:
  409. Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
  410. `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
  411. bucket_size_msec:
  412. Time bucket for aggregation in milliseconds.
  413. filter_by_ts:
  414. List of timestamps to filter the result by specific timestamps.
  415. filter_by_min_value:
  416. Filter result by minimum value (must mention also filter_by_max_value).
  417. filter_by_max_value:
  418. Filter result by maximum value (must mention also filter_by_min_value).
  419. align:
  420. Timestamp for alignment control for aggregation.
  421. latest:
  422. Used when a time series is a compaction, reports the compacted value of the
  423. latest possibly partial bucket
  424. bucket_timestamp:
  425. Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
  426. `high`, `~`, `mid`].
  427. empty:
  428. Reports aggregations for empty buckets.
  429. For more information: https://redis.io/commands/ts.revrange/
  430. """ # noqa
  431. params = self.__range_params(
  432. key,
  433. from_time,
  434. to_time,
  435. count,
  436. aggregation_type,
  437. bucket_size_msec,
  438. filter_by_ts,
  439. filter_by_min_value,
  440. filter_by_max_value,
  441. align,
  442. latest,
  443. bucket_timestamp,
  444. empty,
  445. )
  446. return self.execute_command(REVRANGE_CMD, *params)
  447. def __mrange_params(
  448. self,
  449. aggregation_type: Optional[str],
  450. bucket_size_msec: Optional[int],
  451. count: Optional[int],
  452. filters: List[str],
  453. from_time: Union[int, str],
  454. to_time: Union[int, str],
  455. with_labels: Optional[bool],
  456. filter_by_ts: Optional[List[int]],
  457. filter_by_min_value: Optional[int],
  458. filter_by_max_value: Optional[int],
  459. groupby: Optional[str],
  460. reduce: Optional[str],
  461. select_labels: Optional[List[str]],
  462. align: Optional[Union[int, str]],
  463. latest: Optional[bool],
  464. bucket_timestamp: Optional[str],
  465. empty: Optional[bool],
  466. ):
  467. """Create TS.MRANGE and TS.MREVRANGE arguments."""
  468. params = [from_time, to_time]
  469. self._append_latest(params, latest)
  470. self._append_filer_by_ts(params, filter_by_ts)
  471. self._append_filer_by_value(params, filter_by_min_value, filter_by_max_value)
  472. self._append_with_labels(params, with_labels, select_labels)
  473. self._append_count(params, count)
  474. self._append_align(params, align)
  475. self._append_aggregation(params, aggregation_type, bucket_size_msec)
  476. self._append_bucket_timestamp(params, bucket_timestamp)
  477. self._append_empty(params, empty)
  478. params.extend(["FILTER"])
  479. params += filters
  480. self._append_groupby_reduce(params, groupby, reduce)
  481. return params
  482. def mrange(
  483. self,
  484. from_time: Union[int, str],
  485. to_time: Union[int, str],
  486. filters: List[str],
  487. count: Optional[int] = None,
  488. aggregation_type: Optional[str] = None,
  489. bucket_size_msec: Optional[int] = 0,
  490. with_labels: Optional[bool] = False,
  491. filter_by_ts: Optional[List[int]] = None,
  492. filter_by_min_value: Optional[int] = None,
  493. filter_by_max_value: Optional[int] = None,
  494. groupby: Optional[str] = None,
  495. reduce: Optional[str] = None,
  496. select_labels: Optional[List[str]] = None,
  497. align: Optional[Union[int, str]] = None,
  498. latest: Optional[bool] = False,
  499. bucket_timestamp: Optional[str] = None,
  500. empty: Optional[bool] = False,
  501. ):
  502. """
  503. Query a range across multiple time-series by filters in forward direction.
  504. Args:
  505. from_time:
  506. Start timestamp for the range query. `-` can be used to express the minimum possible timestamp (0).
  507. to_time:
  508. End timestamp for range query, `+` can be used to express the maximum possible timestamp.
  509. filters:
  510. filter to match the time-series labels.
  511. count:
  512. Limits the number of returned samples.
  513. aggregation_type:
  514. Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
  515. `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
  516. bucket_size_msec:
  517. Time bucket for aggregation in milliseconds.
  518. with_labels:
  519. Include in the reply all label-value pairs representing metadata labels of the time series.
  520. filter_by_ts:
  521. List of timestamps to filter the result by specific timestamps.
  522. filter_by_min_value:
  523. Filter result by minimum value (must mention also filter_by_max_value).
  524. filter_by_max_value:
  525. Filter result by maximum value (must mention also filter_by_min_value).
  526. groupby:
  527. Grouping by fields the results (must mention also reduce).
  528. reduce:
  529. Applying reducer functions on each group. Can be one of [`avg` `sum`, `min`,
  530. `max`, `range`, `count`, `std.p`, `std.s`, `var.p`, `var.s`].
  531. select_labels:
  532. Include in the reply only a subset of the key-value pair labels of a series.
  533. align:
  534. Timestamp for alignment control for aggregation.
  535. latest:
  536. Used when a time series is a compaction, reports the compacted
  537. value of the latest possibly partial bucket
  538. bucket_timestamp:
  539. Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
  540. `high`, `~`, `mid`].
  541. empty:
  542. Reports aggregations for empty buckets.
  543. For more information: https://redis.io/commands/ts.mrange/
  544. """ # noqa
  545. params = self.__mrange_params(
  546. aggregation_type,
  547. bucket_size_msec,
  548. count,
  549. filters,
  550. from_time,
  551. to_time,
  552. with_labels,
  553. filter_by_ts,
  554. filter_by_min_value,
  555. filter_by_max_value,
  556. groupby,
  557. reduce,
  558. select_labels,
  559. align,
  560. latest,
  561. bucket_timestamp,
  562. empty,
  563. )
  564. return self.execute_command(MRANGE_CMD, *params)
  565. def mrevrange(
  566. self,
  567. from_time: Union[int, str],
  568. to_time: Union[int, str],
  569. filters: List[str],
  570. count: Optional[int] = None,
  571. aggregation_type: Optional[str] = None,
  572. bucket_size_msec: Optional[int] = 0,
  573. with_labels: Optional[bool] = False,
  574. filter_by_ts: Optional[List[int]] = None,
  575. filter_by_min_value: Optional[int] = None,
  576. filter_by_max_value: Optional[int] = None,
  577. groupby: Optional[str] = None,
  578. reduce: Optional[str] = None,
  579. select_labels: Optional[List[str]] = None,
  580. align: Optional[Union[int, str]] = None,
  581. latest: Optional[bool] = False,
  582. bucket_timestamp: Optional[str] = None,
  583. empty: Optional[bool] = False,
  584. ):
  585. """
  586. Query a range across multiple time-series by filters in reverse direction.
  587. Args:
  588. from_time:
  589. Start timestamp for the range query. - can be used to express the minimum possible timestamp (0).
  590. to_time:
  591. End timestamp for range query, + can be used to express the maximum possible timestamp.
  592. filters:
  593. Filter to match the time-series labels.
  594. count:
  595. Limits the number of returned samples.
  596. aggregation_type:
  597. Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
  598. `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
  599. bucket_size_msec:
  600. Time bucket for aggregation in milliseconds.
  601. with_labels:
  602. Include in the reply all label-value pairs representing metadata labels of the time series.
  603. filter_by_ts:
  604. List of timestamps to filter the result by specific timestamps.
  605. filter_by_min_value:
  606. Filter result by minimum value (must mention also filter_by_max_value).
  607. filter_by_max_value:
  608. Filter result by maximum value (must mention also filter_by_min_value).
  609. groupby:
  610. Grouping by fields the results (must mention also reduce).
  611. reduce:
  612. Applying reducer functions on each group. Can be one of [`avg` `sum`, `min`,
  613. `max`, `range`, `count`, `std.p`, `std.s`, `var.p`, `var.s`].
  614. select_labels:
  615. Include in the reply only a subset of the key-value pair labels of a series.
  616. align:
  617. Timestamp for alignment control for aggregation.
  618. latest:
  619. Used when a time series is a compaction, reports the compacted
  620. value of the latest possibly partial bucket
  621. bucket_timestamp:
  622. Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
  623. `high`, `~`, `mid`].
  624. empty:
  625. Reports aggregations for empty buckets.
  626. For more information: https://redis.io/commands/ts.mrevrange/
  627. """ # noqa
  628. params = self.__mrange_params(
  629. aggregation_type,
  630. bucket_size_msec,
  631. count,
  632. filters,
  633. from_time,
  634. to_time,
  635. with_labels,
  636. filter_by_ts,
  637. filter_by_min_value,
  638. filter_by_max_value,
  639. groupby,
  640. reduce,
  641. select_labels,
  642. align,
  643. latest,
  644. bucket_timestamp,
  645. empty,
  646. )
  647. return self.execute_command(MREVRANGE_CMD, *params)
  648. def get(self, key: KeyT, latest: Optional[bool] = False):
  649. """# noqa
  650. Get the last sample of `key`.
  651. `latest` used when a time series is a compaction, reports the compacted
  652. value of the latest (possibly partial) bucket
  653. For more information: https://redis.io/commands/ts.get/
  654. """ # noqa
  655. params = [key]
  656. self._append_latest(params, latest)
  657. return self.execute_command(GET_CMD, *params)
  658. def mget(
  659. self,
  660. filters: List[str],
  661. with_labels: Optional[bool] = False,
  662. select_labels: Optional[List[str]] = None,
  663. latest: Optional[bool] = False,
  664. ):
  665. """# noqa
  666. Get the last samples matching the specific `filter`.
  667. Args:
  668. filters:
  669. Filter to match the time-series labels.
  670. with_labels:
  671. Include in the reply all label-value pairs representing metadata
  672. labels of the time series.
  673. select_labels:
  674. Include in the reply only a subset of the key-value pair labels of a series.
  675. latest:
  676. Used when a time series is a compaction, reports the compacted
  677. value of the latest possibly partial bucket
  678. For more information: https://redis.io/commands/ts.mget/
  679. """ # noqa
  680. params = []
  681. self._append_latest(params, latest)
  682. self._append_with_labels(params, with_labels, select_labels)
  683. params.extend(["FILTER"])
  684. params += filters
  685. return self.execute_command(MGET_CMD, *params)
  686. def info(self, key: KeyT):
  687. """# noqa
  688. Get information of `key`.
  689. For more information: https://redis.io/commands/ts.info/
  690. """ # noqa
  691. return self.execute_command(INFO_CMD, key)
  692. def queryindex(self, filters: List[str]):
  693. """# noqa
  694. Get all time series keys matching the `filter` list.
  695. For more information: https://redis.io/commands/ts.queryindex/
  696. """ # noq
  697. return self.execute_command(QUERYINDEX_CMD, *filters)
  698. @staticmethod
  699. def _append_uncompressed(params: List[str], uncompressed: Optional[bool]):
  700. """Append UNCOMPRESSED tag to params."""
  701. if uncompressed:
  702. params.extend(["UNCOMPRESSED"])
  703. @staticmethod
  704. def _append_with_labels(
  705. params: List[str],
  706. with_labels: Optional[bool],
  707. select_labels: Optional[List[str]],
  708. ):
  709. """Append labels behavior to params."""
  710. if with_labels and select_labels:
  711. raise DataError(
  712. "with_labels and select_labels cannot be provided together."
  713. )
  714. if with_labels:
  715. params.extend(["WITHLABELS"])
  716. if select_labels:
  717. params.extend(["SELECTED_LABELS", *select_labels])
  718. @staticmethod
  719. def _append_groupby_reduce(
  720. params: List[str], groupby: Optional[str], reduce: Optional[str]
  721. ):
  722. """Append GROUPBY REDUCE property to params."""
  723. if groupby is not None and reduce is not None:
  724. params.extend(["GROUPBY", groupby, "REDUCE", reduce.upper()])
  725. @staticmethod
  726. def _append_retention(params: List[str], retention: Optional[int]):
  727. """Append RETENTION property to params."""
  728. if retention is not None:
  729. params.extend(["RETENTION", retention])
  730. @staticmethod
  731. def _append_labels(params: List[str], labels: Optional[List[str]]):
  732. """Append LABELS property to params."""
  733. if labels:
  734. params.append("LABELS")
  735. for k, v in labels.items():
  736. params.extend([k, v])
  737. @staticmethod
  738. def _append_count(params: List[str], count: Optional[int]):
  739. """Append COUNT property to params."""
  740. if count is not None:
  741. params.extend(["COUNT", count])
  742. @staticmethod
  743. def _append_timestamp(params: List[str], timestamp: Optional[int]):
  744. """Append TIMESTAMP property to params."""
  745. if timestamp is not None:
  746. params.extend(["TIMESTAMP", timestamp])
  747. @staticmethod
  748. def _append_align(params: List[str], align: Optional[Union[int, str]]):
  749. """Append ALIGN property to params."""
  750. if align is not None:
  751. params.extend(["ALIGN", align])
  752. @staticmethod
  753. def _append_aggregation(
  754. params: List[str],
  755. aggregation_type: Optional[str],
  756. bucket_size_msec: Optional[int],
  757. ):
  758. """Append AGGREGATION property to params."""
  759. if aggregation_type is not None:
  760. params.extend(["AGGREGATION", aggregation_type, bucket_size_msec])
  761. @staticmethod
  762. def _append_chunk_size(params: List[str], chunk_size: Optional[int]):
  763. """Append CHUNK_SIZE property to params."""
  764. if chunk_size is not None:
  765. params.extend(["CHUNK_SIZE", chunk_size])
  766. @staticmethod
  767. def _append_duplicate_policy(
  768. params: List[str], command: Optional[str], duplicate_policy: Optional[str]
  769. ):
  770. """Append DUPLICATE_POLICY property to params on CREATE
  771. and ON_DUPLICATE on ADD.
  772. """
  773. if duplicate_policy is not None:
  774. if command == "TS.ADD":
  775. params.extend(["ON_DUPLICATE", duplicate_policy])
  776. else:
  777. params.extend(["DUPLICATE_POLICY", duplicate_policy])
  778. @staticmethod
  779. def _append_filer_by_ts(params: List[str], ts_list: Optional[List[int]]):
  780. """Append FILTER_BY_TS property to params."""
  781. if ts_list is not None:
  782. params.extend(["FILTER_BY_TS", *ts_list])
  783. @staticmethod
  784. def _append_filer_by_value(
  785. params: List[str], min_value: Optional[int], max_value: Optional[int]
  786. ):
  787. """Append FILTER_BY_VALUE property to params."""
  788. if min_value is not None and max_value is not None:
  789. params.extend(["FILTER_BY_VALUE", min_value, max_value])
  790. @staticmethod
  791. def _append_latest(params: List[str], latest: Optional[bool]):
  792. """Append LATEST property to params."""
  793. if latest:
  794. params.append("LATEST")
  795. @staticmethod
  796. def _append_bucket_timestamp(params: List[str], bucket_timestamp: Optional[str]):
  797. """Append BUCKET_TIMESTAMP property to params."""
  798. if bucket_timestamp is not None:
  799. params.extend(["BUCKETTIMESTAMP", bucket_timestamp])
  800. @staticmethod
  801. def _append_empty(params: List[str], empty: Optional[bool]):
  802. """Append EMPTY property to params."""
  803. if empty:
  804. params.append("EMPTY")