Stateful.py 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116
  1. import Alt
  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. input("error 0: could not found correct config file")
  19. exit()
  20. return doc
  21. except:
  22. input("error 0: could not found correct config file")
  23. exit()
  24. def matchTomlKey(tomlName, key, table=None) -> str:
  25. doc = getTomlDoc(tomlName)
  26. if table == None:
  27. return str(doc.item(key))
  28. elif table != None:
  29. d = doc.unwrap()
  30. return str(d[table][key])
  31. def matchTomlKeys(tomlName, keys, table=None) -> list:
  32. doc = getTomlDoc(tomlName)
  33. if table == None:
  34. # rl == 'r'eturn 'l'ist
  35. rl = []
  36. for key in keys:
  37. try:
  38. rl.append(doc.item(key))
  39. except:
  40. pass
  41. return rl
  42. elif table != None:
  43. rl = []
  44. d = doc.unwrap()
  45. for key in keys:
  46. rl.append(d[table][key])
  47. return rl
  48. # Sqlite3
  49. def operate_sqlite3(dbPath, command):
  50. con = sqlite3.connect(dbPath)
  51. cur = con.cursor()
  52. #
  53. cur.execute()
  54. cur.close()
  55. # Markdown
  56. # csv
  57. # MongoDB
  58. # ----- Transit Command Handler -----
  59. def TransitHandler(transit_command):
  60. dbType = transit_command[-1]
  61. dbPath = transit_command[-2]
  62. exec_command = transit_command[0:-3]
  63. if dbType == "sqlite3":
  64. operate_sqlite3(dbPath, exec_command)
  65. elif dbType == "csv":
  66. pass
  67. elif dbType == "mongodb":
  68. pass
  69. elif dbType == "toml":
  70. pass
  71. elif dbType == "md":
  72. pass
  73. else:
  74. input("error 1: could not found correct Data Base")
  75. exit()