Stateful.py 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114
  1. import Exceptions
  2. import base64, datetime
  3. import tomlkit, sqlite3
  4. # kw == 'k'ey 'w'ord; dt_now = date&&time in UTC
  5. # Please consider adding more salt to prevent uid duplication(specially under multi-user use case)
  6. def generate_uid(kw):
  7. dt_now = str(datetime.datetime.now(datetime.timezone.utc))
  8. genText = str.encode(kw + "_" + dt_now)
  9. uid = base64.b64encode(genText)
  10. return(uid)
  11. # ----- Methods binding command to model -----
  12. # Toml
  13. def getTomlDoc(tomlName):
  14. try:
  15. with open(tomlName, "rb") as t:
  16. doc = tomlkit.load(t)
  17. if doc == {}:
  18. Exceptions.err0()
  19. return doc
  20. except:
  21. Exceptions.err0()
  22. def matchTomlKey(tomlName, key, table=None) -> str:
  23. doc = getTomlDoc(tomlName)
  24. if table == None:
  25. return str(doc.item(key))
  26. elif table != None:
  27. d = doc.unwrap()
  28. return str(d[table][key])
  29. def matchTomlKeys(tomlName, keys, table=None) -> list:
  30. doc = getTomlDoc(tomlName)
  31. if table == None:
  32. # rl == 'r'eturn 'l'ist
  33. rl = []
  34. for key in keys:
  35. try:
  36. rl.append(doc.item(key))
  37. except:
  38. pass
  39. return rl
  40. elif table != None:
  41. rl = []
  42. d = doc.unwrap()
  43. for key in keys:
  44. rl.append(d[table][key])
  45. return rl
  46. # Sqlite3
  47. def operate_sqlite3(dbPath, command):
  48. con = sqlite3.connect(dbPath)
  49. cur = con.cursor()
  50. #
  51. cur.execute()
  52. cur.close()
  53. # Markdown
  54. # csv
  55. # MongoDB
  56. # ----- Transit Command Handler -----
  57. def TransitHandler(transit_command):
  58. dbType = transit_command[-1]
  59. dbPath = transit_command[-2]
  60. exec_command = transit_command[0:-3]
  61. if dbType == "sqlite3":
  62. operate_sqlite3(dbPath, exec_command)
  63. elif dbType == "csv":
  64. pass
  65. elif dbType == "mongodb":
  66. pass
  67. elif dbType == "toml":
  68. pass
  69. elif dbType == "md":
  70. pass
  71. else:
  72. input("err 1: correct db not found")
  73. exit()