fork(1) download
  1. import sqlite3
  2. import decimal
  3.  
  4. class DecimalSum:
  5. def __init__(self):
  6. self.sum = None
  7.  
  8. def step(self, value):
  9. if value is None:
  10. return
  11. v = decimal.Decimal(value)
  12. if self.sum is None:
  13. self.sum = v
  14. else:
  15. self.sum += v
  16.  
  17. def finalize(self):
  18. return None if self.sum is None else str(self.sum)
  19.  
  20. def adapt_decimal(d):
  21. return str(d)
  22.  
  23. def convert_decimal(s):
  24. return decimal.Decimal(s.decode('ascii')) # required for Python 3
  25.  
  26. # Register the adapter
  27. sqlite3.register_adapter(decimal.Decimal, adapt_decimal)
  28. # Register the converter
  29. sqlite3.register_converter("DECTEXT", convert_decimal)
  30.  
  31. # We need also sqlite3.PARSE_COLNAMES for returning correct type for
  32. # aggregate function decimal_sum:
  33. con = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)
  34. con.create_aggregate("decimal_sum", 1, DecimalSum)
  35. cur = con.cursor()
  36. cur.execute("""CREATE TABLE a_test(
  37. a_decimal DECTEXT NOT NULL -- will be stored as TEXT
  38. )""")
  39.  
  40. con.execute("INSERT INTO a_test VALUES(?)", (decimal.Decimal("102.20"),))
  41. # Convert sum to decimal explicitly:
  42. cur.execute("SELECT decimal_sum(a_decimal) AS `s [DECTEXT]` from a_test")
  43. data = cur.fetchone()[0]
  44. print(data)
  45. print(type(data))
  46.  
  47. cur.close()
  48. con.close()
Success #stdin #stdout 0.03s 10968KB
stdin
Standard input is empty
stdout
102.20
<class 'decimal.Decimal'>