core.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197
  1. import typer
  2. import enum
  3. import pathlib
  4. import getpass
  5. import json
  6. import os
  7. import uvicorn
  8. from assets import *
  9. default_config = pathlib.Path("config.json")
  10. default_users_db = pathlib.Path("users.json")
  11. default_db = pathlib.Path("database.db")
  12. config_help = "This is configuration file of the app to use."
  13. users_db_help = "This is location of the users json database."
  14. db_help = "This is location of SQLite3 database file."
  15. description = "This is core reservationer package. It could be used to host "
  16. description = description + "app, or manage configuration or database."
  17. app = typer.Typer(help = description)
  18. class user_command(str, enum.Enum):
  19. """
  20. That commands could be used in the user subcommand.
  21. """
  22. add = "register"
  23. delete = "delete"
  24. password = "password-change"
  25. logout = "full-logout"
  26. def password_prompt() -> str:
  27. while True:
  28. first = getpass.getpass("Password: ")
  29. second = getpass.getpass("Repeat password: ")
  30. if first == second:
  31. return first
  32. print("Passwords do not match.")
  33. @app.command()
  34. def server(
  35. port: int = typer.Option(8080, help = "Port to listen on."),
  36. address: str = typer.Option("0.0.0.0", help = "Address to listen on."),
  37. config: pathlib.Path = typer.Option(default_config, help = config_help)
  38. ) -> None:
  39. """
  40. Start app on selected port and interfaces.
  41. """
  42. os.environ["config_file"] = str(config)
  43. server_config = uvicorn.Config(
  44. "assets.server:instance",
  45. port = port,
  46. host = address,
  47. log_level = "info",
  48. proxy_headers = True,
  49. forwarded_allow_ips = "*"
  50. )
  51. app = uvicorn.Server(server_config)
  52. try:
  53. app.run()
  54. except json.JSONDecodeError as error:
  55. print("Can not parse config and user database.")
  56. print(str(error))
  57. @app.command()
  58. def user(
  59. command: user_command = typer.Argument(help = "Command to run on users."),
  60. nick: str = typer.Argument(help = "Nick of the user to work on."),
  61. config: pathlib.Path = typer.Option(default_config, help = config_help)
  62. ) -> None:
  63. """
  64. Modify user database.
  65. """
  66. try:
  67. loader = config_loader(app_config).load(config)
  68. collection = loader.resources.users
  69. # User adding
  70. if command == user_command.add:
  71. # Check that user exists
  72. if collection.exists(nick):
  73. raise Exception("User with that nick already exists.")
  74. # Create new user
  75. creator = user_factory()
  76. creator.nick = nick
  77. creator.password = password_prompt()
  78. # Add it
  79. collection.add(creator.result)
  80. print("Adding \"" + nick + "\" to the database.")
  81. # User remove
  82. elif command == user_command.delete:
  83. # Load from database
  84. target = collection.get_by_nick(nick)
  85. # Check that user exists
  86. if target is None:
  87. raise Exception("User with given nick not exists.")
  88. # When exists remove it
  89. collection.remove(target)
  90. # Change user password
  91. elif command == user_command.password:
  92. # Load user by nick
  93. target = collection.get_by_nick(nick)
  94. # Check that user exists
  95. if target is None:
  96. raise Exception("User not exists, can not change password.")
  97. # Change password
  98. handler = user_factory(target)
  99. handler.password = password_prompt()
  100. modified = handler.result
  101. # Store it in collection
  102. collection.remove(target).add(modified)
  103. # Logout user from all devices
  104. elif command == user_command.logout:
  105. # Load user from database
  106. target = collection.get_by_nick(nick)
  107. # Check that exists
  108. if target is None:
  109. raise Exception("User not exists, can not logout.")
  110. # Refresh apikey
  111. modified = user_factory(target) \
  112. .refresh_apikey() \
  113. .result
  114. # Store result
  115. collection.remove(target).add(modified)
  116. # Save collection to file
  117. users_saver(collection) \
  118. .drop(loader.result.users_path) \
  119. .save()
  120. print("Users database saved success.")
  121. except validator_exception as error:
  122. print("Password is not correct, too easy to break.")
  123. except json.JSONDecodeError as error:
  124. print("User JSON has syntax exception.")
  125. print(str(error))
  126. except Exception as error:
  127. print("Can not done work.")
  128. print(str(error))
  129. @app.command()
  130. def initialize(
  131. config: pathlib.Path = typer.Option(default_config, help = config_help),
  132. users: pathlib.Path = typer.Option(default_users_db, help = users_db_help),
  133. database: pathlib.Path = typer.Option(default_db, help = db_help)
  134. ) -> None:
  135. """
  136. Initialize app configuration.
  137. """
  138. try:
  139. # Generating config file
  140. config_generator(app_config) \
  141. .modify("users_file", str(users)) \
  142. .modify("database_uri", "sqlite:///" + str(database)) \
  143. .save(config)
  144. # Generating new blank users database
  145. users_saver(users_collection()).save(users)
  146. print("Config file is being created.")
  147. except Exception as error:
  148. print("Config initialization failed.")
  149. print(str(error))
  150. if __name__ == "__main__":
  151. app()