Create transcribee-admin cli

backend/pyproject.toml

@@ -46,6 +46,7 @@ notebooks = [
 
 [project.scripts]
 transcribee-migrate = "transcribee_backend.db.run_migrations:main"
+transcribee-admin = "transcribee_backend.admin_cli:main"
 
 [tool.uv]
 override-dependencies = [
@@ -69,8 +70,8 @@ generate_openapi = "python -m scripts.generate_openapi"
 test = "pytest tests/"
 pyright = "pyright transcribee_backend/"
 
-[tool.setuptools]
-packages = ["transcribee_backend", "transcribee_backend.db"]
+[tool.setuptools.packages.find]
+include = ["transcribee_backend*"]
 
 [build-system]
 requires = ["setuptools", "setuptools-scm"]

backend/transcribee_backend/admin_cli/__init__.py

@@ -0,0 +1,40 @@
+import argparse
+
+from .command import Command
+from .commands.create_api_token import CreateApiTokenCmd
+from .commands.create_user import CreateUserCmd
+from .commands.create_user_token import CreateUserTokenCmd
+from .commands.create_worker import CreateWorkerCmd
+from .commands.reset_task import ResetTaskCmd
+from .commands.set_password import SetPasswordCmd
+
+parser = argparse.ArgumentParser()
+subparsers = parser.add_subparsers(metavar="COMMAND")
+
+
+def add_command(name: str, description: str, command: Command):
+    subparser = subparsers.add_parser(name, help=description, description=description)
+    command.configure_parser(subparser)
+    subparser.set_defaults(func=command.run)
+
+
+# all commands belong here
+add_command("create_api_token", "Create an API token", CreateApiTokenCmd())
+add_command("create_user_token", "Create an user token", CreateUserTokenCmd())
+add_command("create_user", "Create a new user", CreateUserCmd())
+add_command("create_worker", "Register a new worker", CreateWorkerCmd())
+add_command("reset_task", "Reset a task", ResetTaskCmd())
+add_command("set_password", "Set the password of a user", SetPasswordCmd())
+
+
+def main():
+    args = parser.parse_args()
+
+    if "func" in args:
+        args.func(args)
+    else:
+        parser.print_help()
+
+
+if __name__ == "__main__":
+    main()

backend/transcribee_backend/admin_cli/command.py

@@ -0,0 +1,15 @@
+from abc import ABC, abstractmethod
+from argparse import ArgumentParser
+
+
+class Command(ABC):
+    def __init__(self):
+        pass
+
+    @abstractmethod
+    def configure_parser(self, parser: ArgumentParser):
+        pass
+
+    @abstractmethod
+    def run(self, args):
+        pass

backend/transcribee_backend/admin_cli/commands/create_api_token.py

@@ -0,0 +1,15 @@
+from transcribee_backend.admin_cli.command import Command
+from transcribee_backend.auth import create_api_token
+from transcribee_backend.db import SessionContextManager
+
+
+class CreateApiTokenCmd(Command):
+    def configure_parser(self, parser):
+        parser.add_argument("--name", required=True)
+
+    def run(self, args):
+        with SessionContextManager(
+            path="management_command:create_api_token"
+        ) as session:
+            token = create_api_token(session=session, name=args.name)
+            print(f"Token created: {token.token}")

backend/transcribee_backend/admin_cli/commands/create_user.py

@@ -0,0 +1,38 @@
+import random
+import string
+
+from transcribee_backend.admin_cli.command import Command
+from transcribee_backend.auth import create_user
+from transcribee_backend.db import SessionContextManager
+from transcribee_backend.exceptions import UserAlreadyExists
+
+
+def random_password():
+    chars = string.ascii_lowercase + string.digits
+    return "".join(random.choice(chars) for _ in range(20))
+
+
+class CreateUserCmd(Command):
+    def configure_parser(self, parser):
+        parser.add_argument("--user", required=True)
+        parser.add_argument("--password", required=False)
+
+    def run(self, args):
+        password = args.password
+
+        if not password:
+            password = random_password()
+
+            print("Auto-generated password.")
+            print("Infos to send to user:")
+            print()
+            print(f"Username: {args.user}")
+            print(f"Password: {password} (Please change)")
+            print()
+
+        with SessionContextManager(path="management_command:create_user") as session:
+            try:
+                create_user(session=session, username=args.user, password=password)
+                print("User created")
+            except UserAlreadyExists:
+                print("Could not create user. A user with that name already exists.")

backend/transcribee_backend/admin_cli/commands/create_user_token.py

@@ -0,0 +1,40 @@
+import datetime
+
+from sqlmodel import select
+from transcribee_backend.admin_cli.command import Command
+from transcribee_backend.auth import generate_user_token
+from transcribee_backend.db import SessionContextManager
+from transcribee_backend.helpers.time import now_tz_aware
+from transcribee_backend.models.user import User
+
+
+class CreateUserTokenCmd(Command):
+    def configure_parser(self, parser):
+        parser.add_argument("--username", required=True)
+        parser.add_argument("--valid-days", required=True)
+
+    def run(self, args):
+        with SessionContextManager(
+            path="management_command:create_user_token"
+        ) as session:
+            valid_days = int(args.valid_days)
+            if valid_days < 0:
+                print("Valid days must be positive")
+                exit(1)
+
+            valid_until = now_tz_aware() + datetime.timedelta(days=valid_days)
+
+            user = session.exec(
+                select(User).where(User.username == args.username)
+            ).one_or_none()
+
+            if user is None:
+                print(f"User {args.user} not found")
+                exit(1)
+
+            key, user_token = generate_user_token(user, valid_until=valid_until)
+            session.add(user_token)
+            session.commit()
+
+            print(f"User token created and valid until {valid_until}")
+            print(f"Secret: {key}")

backend/transcribee_backend/admin_cli/commands/create_worker.py

@@ -0,0 +1,32 @@
+from sqlmodel import select
+from transcribee_backend import utils
+from transcribee_backend.admin_cli.command import Command
+from transcribee_backend.db import SessionContextManager
+from transcribee_backend.models import Worker
+
+
+class CreateWorkerCmd(Command):
+    def configure_parser(self, parser):
+        parser.add_argument("--name", required=True)
+        parser.add_argument("--token", required=False)
+
+    def run(self, args):
+        if args.token is None:
+            args.token = utils.get_random_string()
+
+        with SessionContextManager(path="management_command:create_worker") as session:
+            statement = select(Worker).where(Worker.token == args.token)
+            results = session.exec(statement)
+            existing_worker = results.one_or_none()
+            if existing_worker is None:
+                worker = Worker(
+                    name=args.name,
+                    token=args.token,
+                    last_seen=None,
+                    deactivated_at=None,
+                )
+                session.add(worker)
+                session.commit()
+                print(f"Worker with token {args.token} created")
+            else:
+                print(f"Worker with token {args.token} already exists")