| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197 | 
							- import typer
 
- import enum
 
- import pathlib
 
- import getpass
 
- import json
 
- import os
 
- import uvicorn
 
- from assets import *
 
- default_config = pathlib.Path("config.json")
 
- default_users_db = pathlib.Path("users.json")
 
- default_db = pathlib.Path("database.db")
 
- config_help = "This is configuration file of the app to use."
 
- users_db_help = "This is location of the users json database."
 
- db_help = "This is location of SQLite3 database file."
 
- description = "This is core reservationer package. It could be used to host "
 
- description = description + "app, or manage configuration or database."
 
- app = typer.Typer(help = description)
 
- class user_command(str, enum.Enum):
 
-     """
 
-     That commands could be used in the user subcommand.
 
-     """
 
-     add = "register"
 
-     delete = "delete"
 
-     password = "password-change"
 
-     logout = "full-logout"
 
- def password_prompt() -> str:
 
-     while True: 
 
-         first = getpass.getpass("Password: ")
 
-         second = getpass.getpass("Repeat password: ")
 
-         if first == second:
 
-             return first
 
-         print("Passwords do not match.")
 
- @app.command()
 
- def server(
 
-     port: int = typer.Option(8080, help = "Port to listen on."),
 
-     address: str = typer.Option("0.0.0.0", help = "Address to listen on."),
 
-     config: pathlib.Path = typer.Option(default_config, help = config_help)
 
- ) -> None:
 
-     """
 
-     Start app on selected port and interfaces.
 
-     """
 
-     os.environ["config_file"] = str(config)
 
-     server_config = uvicorn.Config(
 
-         "assets.server:instance", 
 
-         port = port, 
 
-         host = address, 
 
-         log_level = "info",
 
-         proxy_headers = True,
 
-         forwarded_allow_ips = "*"
 
-     )
 
-     app = uvicorn.Server(server_config)
 
-     
 
-     try:    
 
-         app.run()
 
-     except json.JSONDecodeError as error:
 
-         print("Can not parse config and user database.")
 
-         print(str(error))
 
- @app.command()
 
- def user(
 
-     command: user_command = typer.Argument(help = "Command to run on users."),
 
-     nick: str = typer.Argument(help = "Nick of the user to work on."),
 
-     config: pathlib.Path = typer.Option(default_config, help = config_help)
 
- ) -> None:
 
-     """
 
-     Modify user database.
 
-     """
 
-     
 
-     try:
 
-         loader = config_loader(app_config).load(config)
 
-         collection = loader.resources.users
 
-         # User adding 
 
-         if command == user_command.add:
 
-             
 
-             # Check that user exists
 
-             if collection.exists(nick):
 
-                 raise Exception("User with that nick already exists.")
 
-             # Create new user
 
-             creator = user_factory()
 
-             creator.nick = nick
 
-             creator.password = password_prompt()
 
-             
 
-             # Add it
 
-             collection.add(creator.result)
 
-             
 
-             print("Adding \"" + nick + "\" to the database.")
 
-         # User remove
 
-         elif command == user_command.delete:
 
-             
 
-             # Load from database
 
-             target = collection.get_by_nick(nick)
 
-             # Check that user exists
 
-             if target is None:
 
-                 raise Exception("User with given nick not exists.")
 
-             
 
-             # When exists remove it
 
-             collection.remove(target)
 
-         # Change user password
 
-         elif command == user_command.password:
 
-                 
 
-             # Load user by nick 
 
-             target = collection.get_by_nick(nick)
 
-             
 
-             # Check that user exists
 
-             if target is None:
 
-                 raise Exception("User not exists, can not change password.")
 
-             # Change password
 
-             handler = user_factory(target)
 
-             handler.password = password_prompt()
 
-             modified = handler.result
 
-             
 
-             # Store it in collection
 
-             collection.remove(target).add(modified)
 
-         
 
-         # Logout user from all devices
 
-         elif command == user_command.logout:
 
-             
 
-             # Load user from database
 
-             target = collection.get_by_nick(nick)
 
-             
 
-             # Check that exists
 
-             if target is None:
 
-                 raise Exception("User not exists, can not logout.")
 
-             # Refresh apikey
 
-             modified = user_factory(target) \
 
-             .refresh_apikey() \
 
-             .result
 
-             
 
-             # Store result
 
-             collection.remove(target).add(modified)
 
-         # Save collection to file
 
-         users_saver(collection) \
 
-         .drop(loader.result.users_path) \
 
-         .save()
 
-         print("Users database saved success.")
 
-     except validator_exception as error:
 
-         print("Password is not correct, too easy to break.")
 
-     except json.JSONDecodeError as error:  
 
-         print("User JSON has syntax exception.")
 
-         print(str(error))
 
-     except Exception as error:
 
-         print("Can not done work.")
 
-         print(str(error))
 
- @app.command()
 
- def initialize(
 
-     config: pathlib.Path = typer.Option(default_config, help = config_help),
 
-     users: pathlib.Path = typer.Option(default_users_db, help = users_db_help),
 
-     database: pathlib.Path = typer.Option(default_db, help = db_help)
 
- ) -> None:
 
-     """
 
-     Initialize app configuration.
 
-     """
 
-     try:
 
-         
 
-         # Generating config file
 
-         config_generator(app_config) \
 
-         .modify("users_file", str(users)) \
 
-         .modify("database_uri", "sqlite:///" + str(database)) \
 
-         .save(config)
 
-         
 
-         # Generating new blank users database
 
-         users_saver(users_collection()).save(users)
 
-         
 
-         print("Config file is being created.")
 
-     
 
-     except Exception as error:
 
-         print("Config initialization failed.")
 
-         print(str(error))
 
- if __name__ == "__main__":
 
-     app()
 
 
  |