aboutsummaryrefslogtreecommitdiff
path: root/fietsboek/scripts/fietsctl.py
blob: 0043862e119de7e99cf2442d4cf64690217d1f7d (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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
55
56
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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
"""Script to do maintenance work on a Fietsboek instance."""
# pylint: disable=too-many-arguments
from typing import Optional

import click
from click_option_group import RequiredMutuallyExclusiveOptionGroup, optgroup
from pyramid.paster import bootstrap, setup_logging
from pyramid.scripting import AppEnvironment
from sqlalchemy import select

from .. import __VERSION__, models
from . import config_option

EXIT_OKAY = 0
EXIT_FAILURE = 1


def setup(config_path: str) -> AppEnvironment:
    """Sets up the logging and app environment for the scripts.

    This - for example - sets up a transaction manager in
    ``setup(...)["request"]``.

    :param config_path: Path to the configuration file.
    :return: The prepared environment.
    """
    setup_logging(config_path)
    return bootstrap(config_path)


@click.group(
    help=__doc__,
    context_settings={"help_option_names": ["-h", "--help"]},
)
@click.version_option(package_name="fietsboek")
def cli():
    """CLI main entry point."""


@cli.command("useradd")
@config_option
@click.option("--email", help="email address of the user", prompt=True)
@click.option("--name", help="name of the user", prompt=True)
@click.option(
    "--password",
    help="password of the user",
    prompt=True,
    hide_input=True,
    confirmation_prompt=True,
)
@click.option("--admin", help="make the new user an admin", is_flag=True)
@click.pass_context
def cmd_useradd(ctx: click.Context, config: str, email: str, name: str, password: str, admin: bool):
    """Create a new user.

    This user creation bypasses the "enable_account_registration" setting. It
    also immediately sets the new user's account to being verified.

    If email, name or password are not given as command line arguments, they
    will be asked for interactively.

    On success, the created user's unique ID will be printed.

    Note that this function does less input validation and should therefore be used with care!
    """
    env = setup(config)

    # The UNIQUE constraint only prevents identical emails from being inserted,
    # but does not take into account the case insensitivity. The least we
    # should do here to not brick log ins for the user is to check if the email
    # already exists.
    query = models.User.query_by_email(email)
    with env["request"].tm:
        result = env["request"].dbsession.execute(query).scalar_one_or_none()
        if result is not None:
            click.echo("Error: The given email already exists!", err=True)
            ctx.exit(EXIT_FAILURE)

    user = models.User(name=name, email=email, is_verified=True, is_admin=admin)
    user.set_password(password)

    with env["request"].tm:
        dbsession = env["request"].dbsession
        dbsession.add(user)
        dbsession.flush()
        user_id = user.id

    click.echo(user_id)


@cli.command("userdel")
@config_option
@click.option("--force", "-f", help="override the safety check", is_flag=True)
@optgroup.group("User selection", cls=RequiredMutuallyExclusiveOptionGroup)
@optgroup.option("--id", "-i", "id_", help="database ID of the user", type=int)
@optgroup.option("--email", "-e", help="email of the user")
@click.pass_context
def cmd_userdel(
    ctx: click.Context,
    config: str,
    force: bool,
    id_: Optional[int],
    email: Optional[str],
):
    """Delete a user.

    This command deletes the user's account as well as any tracks associated
    with it.

    This command is destructive and irreversibly deletes data.
    """
    env = setup(config)
    if id_ is not None:
        query = select(models.User).filter_by(id=id_)
    else:
        query = models.User.query_by_email(email)
    with env["request"].tm:
        dbsession = env["request"].dbsession
        user = dbsession.execute(query).scalar_one_or_none()
        if user is None:
            click.echo("Error: No such user found.", err=True)
            ctx.exit(EXIT_FAILURE)
        click.echo(user.name)
        click.echo(user.email)
        if not force:
            if not click.confirm("Really delete this user?"):
                click.echo("Aborted by user.")
                ctx.exit(EXIT_FAILURE)
        dbsession.delete(user)
        click.echo("User deleted")


@cli.command("userlist")
@config_option
def cmd_userlist(config: str):
    """Prints a listing of all user accounts.

    The format is:

        [av] {ID} - {email} - {name}

    with one line per user. The 'a' is added for admin accounts, the 'v' is added
    for verified users.
    """
    env = setup(config)
    with env["request"].tm:
        dbsession = env["request"].dbsession
        users = dbsession.execute(select(models.User).order_by(models.User.id)).scalars()
        for user in users:
            # pylint: disable=consider-using-f-string
            tag = "[{}{}]".format(
                "a" if user.is_admin else "-",
                "v" if user.is_verified else "-",
            )
            click.echo(f"{tag} {user.id} - {user.email} - {user.name}")


@cli.command("passwd")
@config_option
@click.option("--password", help="password of the user")
@optgroup.group("User selection", cls=RequiredMutuallyExclusiveOptionGroup)
@optgroup.option("--id", "-i", "id_", help="database ID of the user", type=int)
@optgroup.option("--email", "-e", help="email of the user")
@click.pass_context
def cmd_passwd(
    ctx: click.Context,
    config: str,
    password: Optional[str],
    id_: Optional[int],
    email: Optional[str],
):
    """Change the password of a user."""
    env = setup(config)
    if id_ is not None:
        query = select(models.User).filter_by(id=id_)
    else:
        query = models.User.query_by_email(email)
    with env["request"].tm:
        dbsession = env["request"].dbsession
        user = dbsession.execute(query).scalar_one_or_none()
        if user is None:
            click.echo("Error: No such user found.", err=True)
            ctx.exit(EXIT_FAILURE)
        if not password:
            password = click.prompt("Password", hide_input=True, confirmation_prompt=True)

        user.set_password(password)
        click.echo(f"Changed password of {user.name} ({user.email})")


@cli.command("maintenance-mode")
@config_option
@click.option("--disable", help="disable the maintenance mode", is_flag=True)
@click.argument("reason", required=False)
@click.pass_context
def cmd_maintenance_mode(ctx: click.Context, config: str, disable: bool, reason: Optional[str]):
    """Check the status of the maintenance mode, or activate or deactive it.

    When REASON is given, enables the maintenance mode with the given text as
    reason.

    With neither --disable nor REASON given, just checks the state of the
    maintenance mode.
    """
    env = setup(config)
    data_manager = env["request"].data_manager
    if disable and reason:
        click.echo("Cannot enable and disable maintenance mode at the same time", err=True)
        ctx.exit(EXIT_FAILURE)
    elif not disable and not reason:
        maintenance = data_manager.maintenance_mode()
        if maintenance is None:
            click.echo("Maintenance mode is disabled")
        else:
            click.echo(f"Maintenance mode is enabled: {maintenance}")
    elif disable:
        (data_manager.data_dir / "MAINTENANCE").unlink()
    else:
        (data_manager.data_dir / "MAINTENANCE").write_text(reason, encoding="utf-8")


@cli.command("version")
def cmd_version():
    """Show the installed fietsboek version."""
    name = __name__.split(".", 1)[0]
    print(f"{name} {__VERSION__}")