| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197 | import typerimport enumimport pathlibimport getpassimport jsonimport osimport uvicornfrom 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()
 |