Skip to content

CLI

elva.apps.server.cli

CLI definition.

Functions:

  • resolve_persistence –

    Derive path and persistence value from the persistent CLI option.

  • cli –

    Run a WebSocket server.

Attributes:

APP_NAME = 'server' module-attribute

The name of the app.

resolve_persistence(ctx, param, persistent)

Derive path and persistence value from the persistent CLI option.

It sets an additional path parameter in the context.

Parameters:

  • ctx (Context) –

    the context of the current invokation.

  • param (Parameter) –

    the option parameter object.

  • persistent (bool | str) –

    the value passed from the CLI.

Returns:

  • bool –

    the derive persistence flag.

Source code in src/elva/apps/server/cli.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def resolve_persistence(
    ctx: click.Context, param: click.Parameter, persistent: bool | str
) -> bool:
    """
    Derive path and persistence value from the `persistent` CLI option.

    It sets an additional `path` parameter in the context.

    Arguments:
        ctx: the context of the current invokation.
        param: the option parameter object.
        persistent: the value passed from the CLI.

    Returns:
        the derive persistence flag.
    """
    match persistent:
        # no flag given
        case None:
            path = None
            persistent = False
        # flag given, but without a path
        case "":
            path = None
            persistent = True
        # anything else, i.e. a flag given with a path
        case _:
            path = Path(persistent).resolve()

            if path.exists() and not path.is_dir():
                raise click.BadArgumentUsage(
                    f"the given path '{path}' is not a directory"
                )

            persistent = True

    ctx.params["path"] = path

    return persistent

cli(config, *args, **kwargs)

Run a WebSocket server.

Parameters:

  • config (dict) –

    the merged configuration parameters from CLI and files.

  • args (tuple, default: () ) –

    unused positional arguments.

  • kwargs (dict, default: {} ) –

    parameters passed from the CLI.

Source code in src/elva/apps/server/cli.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@click.command(name=APP_NAME)
@common_options
@click.option(
    "--persistent",
    # one needs to set this manually here since one cannot use
    # the keyword argument `type=click.Path(...)` as it would collide
    # with `flag_value=""`
    metavar="[DIRECTORY]",
    help=(
        "Hold the received content in a local YDoc in volatile memory "
        "or also save it under DIRECTORY if given. "
        "Without this flag, the server simply broadcasts all incoming messages "
        "within the respective room."
    ),
    # explicitely stating that the argument to this option is optional
    # see: https://github.com/pallets/click/pull/1618#issue-649167183
    is_flag=False,
    # used when no argument is given to flag
    flag_value="",
    callback=resolve_persistence,
)
@click.option(
    "--ldap",
    metavar="REALM SERVER BASE",
    help="Enable Basic Authentication via LDAP self bind.",
    nargs=3,
    type=str,
)
@click.option(
    "--dummy",
    help="Enable Dummy Basic Authentication. DO NOT USE IN PRODUCTION.",
    is_flag=True,
)
@pass_config_for(APP_NAME)
def cli(config: dict, *args: tuple, **kwargs: dict):
    """
    Run a WebSocket server.
    \f

    Arguments:
        config: the merged configuration parameters from CLI and files.
        args: unused positional arguments.
        kwargs: parameters passed from the CLI.
    """
    # imports
    logging = import_("logging")
    sys = import_("sys")

    anyio = import_("anyio")

    _log = import_("elva.log")
    app = import_("elva.apps.server.app")

    # logging
    _log.LOGGER_NAME.set(__package__)
    log = logging.getLogger(__package__)

    if config.get("log") is not None:
        log_handler = logging.FileHandler(config["log"])
    else:
        log_handler = logging.StreamHandler(sys.stdout)
    log_handler.setFormatter(_log.DefaultFormatter())
    log.addHandler(log_handler)

    level_name = config.get("level") or "INFO"
    level = logging.getLevelNamesMapping()[level_name]
    log.setLevel(level)

    # run app, catch file permission errors with an appropriate message
    try:
        anyio.run(app.main, config)
    except PermissionError as exc:
        raise click.UsageError(exc)
    except KeyboardInterrupt:
        pass