utils.py 1.3 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344
  1. from ..helpers import nativestr
  2. def list_to_dict(aList):
  3. return {nativestr(aList[i][0]): nativestr(aList[i][1]) for i in range(len(aList))}
  4. def parse_range(response):
  5. """Parse range response. Used by TS.RANGE and TS.REVRANGE."""
  6. return [tuple((r[0], float(r[1]))) for r in response]
  7. def parse_m_range(response):
  8. """Parse multi range response. Used by TS.MRANGE and TS.MREVRANGE."""
  9. res = []
  10. for item in response:
  11. res.append({nativestr(item[0]): [list_to_dict(item[1]), parse_range(item[2])]})
  12. return sorted(res, key=lambda d: list(d.keys()))
  13. def parse_get(response):
  14. """Parse get response. Used by TS.GET."""
  15. if not response:
  16. return None
  17. return int(response[0]), float(response[1])
  18. def parse_m_get(response):
  19. """Parse multi get response. Used by TS.MGET."""
  20. res = []
  21. for item in response:
  22. if not item[2]:
  23. res.append({nativestr(item[0]): [list_to_dict(item[1]), None, None]})
  24. else:
  25. res.append(
  26. {
  27. nativestr(item[0]): [
  28. list_to_dict(item[1]),
  29. int(item[2][0]),
  30. float(item[2][1]),
  31. ]
  32. }
  33. )
  34. return sorted(res, key=lambda d: list(d.keys()))