Loading src/jama/adminui/app.py +88 −0 Changes for src/jama/adminui/app.py: 88 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -34,11 +34,13 @@ from jama.admin_services import config as config_services from jama.admin_services import tasks as task_services from resources import acl from resources.admin_services import access as access_services from resources.admin_services import api_keys as api_key_services from resources.admin_services import filetypes as filetype_services from resources.admin_services import metadata as metadata_services from resources.admin_services import projects as project_services from resources.admin_services import users as user_services from resources.models import ( APIKey, FileType, Metadata, MetadataSet, Loading Loading @@ -595,6 +597,7 @@ class JamaAdminApp(App): ("ctrl+r", "roles", "Roles"), ("m", "metadata", "Metadata"), ("ctrl+u", "members", "Members"), ("ctrl+k", "api_keys", "API keys"), ("s", "delete_successful_tasks", "Delete successful"), ("f", "delete_failed_tasks", "Delete failed"), ("space", "toggle", "Toggle"), Loading Loading @@ -670,6 +673,7 @@ class JamaAdminApp(App): ], "users": [ FooterCommand("Enter project roles", "open_selected"), FooterCommand("Ctrl+K API keys", "api_keys"), FooterCommand("A add user", "add"), FooterCommand("E edit user", "edit"), FooterCommand("D delete user", "delete"), Loading @@ -677,12 +681,21 @@ class JamaAdminApp(App): FooterCommand("Q quit", "quit"), ], "user_accesses": [ FooterCommand("Ctrl+K API keys", "api_keys"), FooterCommand("A add project role", "add"), FooterCommand("D remove project role", "delete"), FooterCommand("Esc back", "back"), FooterCommand("R refresh", "refresh"), FooterCommand("Q quit", "quit"), ], "user_api_keys": [ FooterCommand("A add API key", "add"), FooterCommand("E edit API key", "edit"), FooterCommand("D delete API key", "delete"), FooterCommand("Esc back", "back"), FooterCommand("R refresh", "refresh"), FooterCommand("Q quit", "quit"), ], "tasks": [ FooterCommand("Enter details", "open_selected"), FooterCommand("A filter", "add"), Loading Loading @@ -810,6 +823,11 @@ class JamaAdminApp(App): if project: self.open_mode("project_members", {"project_id": project.pk}) def action_api_keys(self) -> None: user = self.selected_user() if user: self.open_mode("user_api_keys", {"user_id": user.pk}) def action_open_selected(self) -> None: handlers = { "projects": self.open_project_stats, Loading Loading @@ -837,6 +855,7 @@ class JamaAdminApp(App): "filetypes": self.add_filetype, "tasks": self.filter_tasks, "cache": self.clear_all_caches, "user_api_keys": self.add_api_key, } handler = handlers.get(self.mode_name) if handler: Loading @@ -852,6 +871,7 @@ class JamaAdminApp(App): "metadata": self.edit_metadata, "filetypes": self.edit_filetype, "config": self.edit_env_entry, "user_api_keys": self.edit_api_key, } handler = handlers.get(self.mode_name) if handler: Loading @@ -867,6 +887,7 @@ class JamaAdminApp(App): "metadata_sets": self.delete_metadata_set, "metadata": self.delete_metadata, "filetypes": self.delete_filetype, "user_api_keys": self.delete_api_key, } handler = handlers.get(self.mode_name) if handler: Loading Loading @@ -979,6 +1000,14 @@ class JamaAdminApp(App): ) return None def selected_user(self) -> User | None: if self.mode_name == "users": selected = self.selected_object() return selected if isinstance(selected, User) else None if self.context.get("user_id"): return User.objects.filter(pk=self.context["user_id"]).first() return None def refresh_table(self) -> None: try: getattr(self, f"load_{self.mode_name}")() Loading Loading @@ -1730,6 +1759,65 @@ class JamaAdminApp(App): ), ) def load_user_api_keys(self) -> None: user = get_object_or_404(User, pk=self.context["user_id"]) self.set_title(f"{self.translate('API keys')}: {user.username}") self.reset_table(["id", "active", "key hash"]) for api_key in api_key_services.list_api_keys(user=user): self.add_row( f"api-key:{api_key.pk}", api_key, [api_key.pk, api_key.active, api_key.key_hash], ) self.set_status("A add API key. E activate/deactivate. D delete.") def add_api_key(self) -> None: user = get_object_or_404(User, pk=self.context["user_id"]) def create(values): result = api_key_services.create_api_key( user, values["key"], active=values["active"] ) if result.generated: return f"New API key (copy it now): {result.raw_key}" return "API key saved." self.form( "Add API key", [ Field("key", "API key (blank = generate)", "", "password"), Field("active", "Active", True, "bool"), ], create, ) def edit_api_key(self) -> None: api_key = self.selected_object() if isinstance(api_key, APIKey): def update(values): api_key_services.update_api_key(api_key, active=values["active"]) return "API key updated." self.form( f"{self.translate('Edit API key')} {api_key.pk}", [Field("active", "Active", api_key.active, "bool")], update, ) def delete_api_key(self) -> None: api_key = self.selected_object() if isinstance(api_key, APIKey): def delete(_): api_key_services.delete_api_key(api_key) return "API key deleted." self.confirm( f"Delete API key {api_key.pk}?", delete, ) def load_tasks(self) -> None: self.set_title("Task results") self.reset_table(["id", "project", "task_path", "enqueued_at", "status"]) Loading src/jama/adminui/i18n.py +13 −0 Changes for src/jama/adminui/i18n.py: 13 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -31,6 +31,7 @@ FR_TRANSLATIONS = { "Ctrl+R roles": "Ctrl+R rôles", "M metadata": "M métadonnées", "Ctrl+U members": "Ctrl+U membres", "Ctrl+K API keys": "Ctrl+K clés API", "R refresh": "R actualiser", "Q quit": "Q quitter", "Esc back": "Esc retour", Loading @@ -44,6 +45,9 @@ FR_TRANSLATIONS = { "D delete scoped": "D supprimer ciblée", "A add member": "A ajouter membre", "D remove selected role": "D retirer rôle sélectionné", "A add API key": "A ajouter clé API", "E edit API key": "E modifier clé API", "D delete API key": "D supprimer clé API", "Enter metadata": "Entrée métadonnées", "A add set": "A ajouter jeu", "E rename set": "E renommer jeu", Loading Loading @@ -111,6 +115,7 @@ FR_TRANSLATIONS = { "iiif": "IIIF", "hls": "HLS", "key": "clé", "key hash": "empreinte de clé", "utility": "utilitaire", "alias": "alias", "line": "ligne", Loading Loading @@ -163,6 +168,10 @@ FR_TRANSLATIONS = { "Last name": "Nom", "Password (blank = unusable)": "Mot de passe (vide = inutilisable)", "Active": "Actif", "API keys": "Clés API", "Add API key": "Ajouter une clé API", "Edit API key": "Modifier la clé API", "API key (blank = generate)": "Clé API (vide = générer)", "Staff": "Équipe", "Superuser": "Superutilisateur", "New password (blank = unchanged)": "Nouveau mot de passe (vide = inchangé)", Loading Loading @@ -214,6 +223,10 @@ FR_TRANSLATIONS = { "A add metadata. E edit. D delete.": "A ajouter métadonnée. E modifier. D supprimer.", "Enter opens project roles. A add, E edit, D delete.": "Entrée ouvre les rôles projet. A ajouter, E modifier, D supprimer.", "A add by project/role ids. D remove selected access.": "A ajoute par projet/rôle. D retire l'accès sélectionné.", "A add API key. E activate/deactivate. D delete.": "A ajoute une clé API. E active/désactive. D supprime.", "API key saved.": "Clé API enregistrée.", "API key updated.": "Clé API mise à jour.", "API key deleted.": "Clé API supprimée.", "A filters. Enter details. S deletes SUCCESSFUL. F deletes FAILED.": "A filtre. Entrée détails. S supprime SUCCESSFUL. F supprime FAILED.", "Esc back.": "Esc retour.", "A add, E edit, D delete. Utilities can load built-ins.": "A ajouter, E modifier, D supprimer. Les utilitaires peuvent charger les types intégrés.", Loading src/jama/tests.py +21 −0 Changes for src/jama/tests.py: 21 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -379,6 +379,27 @@ class AdminUiAppTestCase(TransactionTestCase): with patch.dict(os.environ, {"DJANGO_ALLOW_ASYNC_UNSAFE": "true"}): asyncio.run(run_app()) def test_ctrl_k_on_user_opens_api_keys(self): user = User.objects.create(username="api-key-user") api_key, _ = APIKey.create_for_key(user, "raw-key") async def run_app(): from jama.adminui.app import JamaAdminApp app = JamaAdminApp() async with app.run_test() as pilot: app.load("users") await pilot.pause() await pilot.press("ctrl+k") await pilot.pause() self.assertEqual(app.mode_name, "user_api_keys") self.assertEqual(app.context["user_id"], user.pk) self.assertEqual(app.selected_object().pk, api_key.pk) with patch.dict(os.environ, {"DJANGO_ALLOW_ASYNC_UNSAFE": "true"}): asyncio.run(run_app()) def test_m_on_project_opens_metadata_sets(self): project = Project.objects.create(label="Metadata project", description="") Loading src/resources/admin_services/api_keys.py 0 → 100644 +109 −0 Changes for src/resources/admin_services/api_keys.py: 109 added lines, 0 removed lines. Original line number Diff line number Diff line import secrets import string from dataclasses import dataclass from django.contrib.auth.models import User from resources.models import APIKey @dataclass(frozen=True) class APIKeyCreation: api_key: APIKey raw_key: str created: bool generated: bool def generate_api_key(length: int = 32) -> str: alphabet = string.ascii_letters + string.digits return "".join(secrets.choice(alphabet) for _ in range(length)) def list_api_keys(*, user: User | None = None, active_only: bool = False): api_keys = APIKey.objects.select_related("user").order_by("user__username", "id") if user is not None: api_keys = api_keys.filter(user=user) if active_only: api_keys = api_keys.filter(active=True) return api_keys def create_api_key( user: User, raw_key: str | None = None, *, active: bool = True ) -> APIKeyCreation: generated = not raw_key raw_key = raw_key or generate_api_key() api_key, created = APIKey.create_for_key(user=user, key=raw_key) if api_key.active != active: api_key.active = active api_key.save(update_fields=["active"]) return APIKeyCreation(api_key, raw_key, created, generated) def create_api_key_for_username( username: str, raw_key: str | None = None, *, active: bool = True, create_user: bool = False, ) -> APIKeyCreation: user = User.objects.filter(username=username).first() if user is None: if not create_user: raise ValueError( f'User "{username}" does not exist. Use --create-user to create it.' ) user = User.objects.create(username=username) return create_api_key(user, raw_key, active=active) def update_api_key(api_key: APIKey, *, active: bool) -> APIKey: api_key.active = active api_key.save(update_fields=["active"]) return api_key def get_api_key( *, key_id: int | None = None, key_hash: str | None = None, username: str | None = None, raw_key: str | None = None, ) -> APIKey: selectors = sum( selected for selected in ( key_id is not None, key_hash is not None, username is not None or raw_key is not None, ) ) if selectors != 1: raise ValueError( "Delete exactly one key using --id, --hash, or username plus raw key." ) if (username is None) != (raw_key is None): raise ValueError("Deleting by raw key requires both username and key.") filters = {} if key_id is not None: filters["id"] = key_id elif key_hash is not None: filters["key_hash"] = key_hash else: user = User.objects.filter(username=username).first() if user is None: raise ValueError(f'User "{username}" does not exist.') filters.update(user=user, key_hash=APIKey.hash_key(raw_key)) try: return APIKey.objects.select_related("user").get(**filters) except APIKey.DoesNotExist: raise ValueError("API key does not exist.") from None def delete_api_key(api_key: APIKey) -> tuple[int, str]: deleted_id = api_key.pk username = api_key.user.username api_key.delete() return deleted_id, username src/resources/management/commands/apikey.py +34 −73 Changes for src/resources/management/commands/apikey.py: 34 added lines, 73 removed lines. Original line number Diff line number Diff line import secrets import string from django.contrib.auth.models import User from django.core.management.base import BaseCommand, CommandError from resources.models import APIKey def generate_api_key(length=32): return "".join( secrets.choice(string.ascii_letters + string.digits) for _ in range(length) ) from resources.admin_services import api_keys as api_key_services class Command(BaseCommand): Loading Loading @@ -75,35 +65,34 @@ class Command(BaseCommand): self.handle_delete(options) def handle_add(self, options): username = options["username"] user = User.objects.filter(username=username).first() if user is None: if not options["create_user"]: raise CommandError( f'User "{username}" does not exist. Use --create-user to create it.' ) user = User.objects.create(username=username) raw_key = options["key"] or generate_api_key() api_key, created = APIKey.create_for_key(user=user, key=raw_key) api_key.active = not options["inactive"] api_key.save(update_fields=["active"]) status = "created" if created else "updated" try: result = api_key_services.create_api_key_for_username( options["username"], options["key"], active=not options["inactive"], create_user=options["create_user"], ) except ValueError as error: raise CommandError(str(error)) from error status = "created" if result.created else "updated" self.stdout.write( f"{status}\tid={api_key.pk}\tuser={user.username}\tactive={api_key.active}" f"{status}\tid={result.api_key.pk}\tuser={result.api_key.user.username}" f"\tactive={result.api_key.active}" ) if options["key"] is None: self.stdout.write(f"key\t{raw_key}") if result.generated: self.stdout.write(f"key\t{result.raw_key}") def handle_list(self, options): api_keys = APIKey.objects.select_related("user").order_by( "user__username", "id" ) user = None if options["username"]: api_keys = api_keys.filter(user__username=options["username"]) if options["active_only"]: api_keys = api_keys.filter(active=True) from django.contrib.auth.models import User user = User.objects.filter(username=options["username"]).first() if user is None: return self.stdout.write("id\tuser\tactive\tkey_hash") api_keys = api_key_services.list_api_keys( user=user, active_only=options["active_only"] ) self.stdout.write("id\tuser\tactive\tkey_hash") for api_key in api_keys: Loading @@ -117,42 +106,14 @@ class Command(BaseCommand): username = options["username"] raw_key = options["key"] selectors = sum( 1 for is_selected in ( key_id is not None, key_hash is not None, username is not None or raw_key is not None, ) if is_selected ) if selectors != 1: raise CommandError( "Delete exactly one key using --id, --hash, or username plus raw key." ) if (username is None) != (raw_key is None): raise CommandError("Deleting by raw key requires both username and key.") api_keys = APIKey.objects.select_related("user") if key_id is not None: api_key = self.get_api_key(api_keys, id=key_id) elif key_hash is not None: api_key = self.get_api_key(api_keys, key_hash=key_hash) else: user = User.objects.filter(username=username).first() if user is None: raise CommandError(f'User "{username}" does not exist.') api_key = self.get_api_key( api_keys, user=user, key_hash=APIKey.hash_key(raw_key) ) deleted_id = api_key.pk deleted_user = api_key.user.username api_key.delete() self.stdout.write(f"deleted\tid={deleted_id}\tuser={deleted_user}") def get_api_key(self, api_keys, **filters): try: return api_keys.get(**filters) except APIKey.DoesNotExist: raise CommandError("API key does not exist.") from None api_key = api_key_services.get_api_key( key_id=key_id, key_hash=key_hash, username=username, raw_key=raw_key, ) deleted_id, deleted_user = api_key_services.delete_api_key(api_key) except ValueError as error: raise CommandError(str(error)) from error self.stdout.write(f"deleted\tid={deleted_id}\tuser={deleted_user}") Loading
src/jama/adminui/app.py +88 −0 Changes for src/jama/adminui/app.py: 88 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -34,11 +34,13 @@ from jama.admin_services import config as config_services from jama.admin_services import tasks as task_services from resources import acl from resources.admin_services import access as access_services from resources.admin_services import api_keys as api_key_services from resources.admin_services import filetypes as filetype_services from resources.admin_services import metadata as metadata_services from resources.admin_services import projects as project_services from resources.admin_services import users as user_services from resources.models import ( APIKey, FileType, Metadata, MetadataSet, Loading Loading @@ -595,6 +597,7 @@ class JamaAdminApp(App): ("ctrl+r", "roles", "Roles"), ("m", "metadata", "Metadata"), ("ctrl+u", "members", "Members"), ("ctrl+k", "api_keys", "API keys"), ("s", "delete_successful_tasks", "Delete successful"), ("f", "delete_failed_tasks", "Delete failed"), ("space", "toggle", "Toggle"), Loading Loading @@ -670,6 +673,7 @@ class JamaAdminApp(App): ], "users": [ FooterCommand("Enter project roles", "open_selected"), FooterCommand("Ctrl+K API keys", "api_keys"), FooterCommand("A add user", "add"), FooterCommand("E edit user", "edit"), FooterCommand("D delete user", "delete"), Loading @@ -677,12 +681,21 @@ class JamaAdminApp(App): FooterCommand("Q quit", "quit"), ], "user_accesses": [ FooterCommand("Ctrl+K API keys", "api_keys"), FooterCommand("A add project role", "add"), FooterCommand("D remove project role", "delete"), FooterCommand("Esc back", "back"), FooterCommand("R refresh", "refresh"), FooterCommand("Q quit", "quit"), ], "user_api_keys": [ FooterCommand("A add API key", "add"), FooterCommand("E edit API key", "edit"), FooterCommand("D delete API key", "delete"), FooterCommand("Esc back", "back"), FooterCommand("R refresh", "refresh"), FooterCommand("Q quit", "quit"), ], "tasks": [ FooterCommand("Enter details", "open_selected"), FooterCommand("A filter", "add"), Loading Loading @@ -810,6 +823,11 @@ class JamaAdminApp(App): if project: self.open_mode("project_members", {"project_id": project.pk}) def action_api_keys(self) -> None: user = self.selected_user() if user: self.open_mode("user_api_keys", {"user_id": user.pk}) def action_open_selected(self) -> None: handlers = { "projects": self.open_project_stats, Loading Loading @@ -837,6 +855,7 @@ class JamaAdminApp(App): "filetypes": self.add_filetype, "tasks": self.filter_tasks, "cache": self.clear_all_caches, "user_api_keys": self.add_api_key, } handler = handlers.get(self.mode_name) if handler: Loading @@ -852,6 +871,7 @@ class JamaAdminApp(App): "metadata": self.edit_metadata, "filetypes": self.edit_filetype, "config": self.edit_env_entry, "user_api_keys": self.edit_api_key, } handler = handlers.get(self.mode_name) if handler: Loading @@ -867,6 +887,7 @@ class JamaAdminApp(App): "metadata_sets": self.delete_metadata_set, "metadata": self.delete_metadata, "filetypes": self.delete_filetype, "user_api_keys": self.delete_api_key, } handler = handlers.get(self.mode_name) if handler: Loading Loading @@ -979,6 +1000,14 @@ class JamaAdminApp(App): ) return None def selected_user(self) -> User | None: if self.mode_name == "users": selected = self.selected_object() return selected if isinstance(selected, User) else None if self.context.get("user_id"): return User.objects.filter(pk=self.context["user_id"]).first() return None def refresh_table(self) -> None: try: getattr(self, f"load_{self.mode_name}")() Loading Loading @@ -1730,6 +1759,65 @@ class JamaAdminApp(App): ), ) def load_user_api_keys(self) -> None: user = get_object_or_404(User, pk=self.context["user_id"]) self.set_title(f"{self.translate('API keys')}: {user.username}") self.reset_table(["id", "active", "key hash"]) for api_key in api_key_services.list_api_keys(user=user): self.add_row( f"api-key:{api_key.pk}", api_key, [api_key.pk, api_key.active, api_key.key_hash], ) self.set_status("A add API key. E activate/deactivate. D delete.") def add_api_key(self) -> None: user = get_object_or_404(User, pk=self.context["user_id"]) def create(values): result = api_key_services.create_api_key( user, values["key"], active=values["active"] ) if result.generated: return f"New API key (copy it now): {result.raw_key}" return "API key saved." self.form( "Add API key", [ Field("key", "API key (blank = generate)", "", "password"), Field("active", "Active", True, "bool"), ], create, ) def edit_api_key(self) -> None: api_key = self.selected_object() if isinstance(api_key, APIKey): def update(values): api_key_services.update_api_key(api_key, active=values["active"]) return "API key updated." self.form( f"{self.translate('Edit API key')} {api_key.pk}", [Field("active", "Active", api_key.active, "bool")], update, ) def delete_api_key(self) -> None: api_key = self.selected_object() if isinstance(api_key, APIKey): def delete(_): api_key_services.delete_api_key(api_key) return "API key deleted." self.confirm( f"Delete API key {api_key.pk}?", delete, ) def load_tasks(self) -> None: self.set_title("Task results") self.reset_table(["id", "project", "task_path", "enqueued_at", "status"]) Loading
src/jama/adminui/i18n.py +13 −0 Changes for src/jama/adminui/i18n.py: 13 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -31,6 +31,7 @@ FR_TRANSLATIONS = { "Ctrl+R roles": "Ctrl+R rôles", "M metadata": "M métadonnées", "Ctrl+U members": "Ctrl+U membres", "Ctrl+K API keys": "Ctrl+K clés API", "R refresh": "R actualiser", "Q quit": "Q quitter", "Esc back": "Esc retour", Loading @@ -44,6 +45,9 @@ FR_TRANSLATIONS = { "D delete scoped": "D supprimer ciblée", "A add member": "A ajouter membre", "D remove selected role": "D retirer rôle sélectionné", "A add API key": "A ajouter clé API", "E edit API key": "E modifier clé API", "D delete API key": "D supprimer clé API", "Enter metadata": "Entrée métadonnées", "A add set": "A ajouter jeu", "E rename set": "E renommer jeu", Loading Loading @@ -111,6 +115,7 @@ FR_TRANSLATIONS = { "iiif": "IIIF", "hls": "HLS", "key": "clé", "key hash": "empreinte de clé", "utility": "utilitaire", "alias": "alias", "line": "ligne", Loading Loading @@ -163,6 +168,10 @@ FR_TRANSLATIONS = { "Last name": "Nom", "Password (blank = unusable)": "Mot de passe (vide = inutilisable)", "Active": "Actif", "API keys": "Clés API", "Add API key": "Ajouter une clé API", "Edit API key": "Modifier la clé API", "API key (blank = generate)": "Clé API (vide = générer)", "Staff": "Équipe", "Superuser": "Superutilisateur", "New password (blank = unchanged)": "Nouveau mot de passe (vide = inchangé)", Loading Loading @@ -214,6 +223,10 @@ FR_TRANSLATIONS = { "A add metadata. E edit. D delete.": "A ajouter métadonnée. E modifier. D supprimer.", "Enter opens project roles. A add, E edit, D delete.": "Entrée ouvre les rôles projet. A ajouter, E modifier, D supprimer.", "A add by project/role ids. D remove selected access.": "A ajoute par projet/rôle. D retire l'accès sélectionné.", "A add API key. E activate/deactivate. D delete.": "A ajoute une clé API. E active/désactive. D supprime.", "API key saved.": "Clé API enregistrée.", "API key updated.": "Clé API mise à jour.", "API key deleted.": "Clé API supprimée.", "A filters. Enter details. S deletes SUCCESSFUL. F deletes FAILED.": "A filtre. Entrée détails. S supprime SUCCESSFUL. F supprime FAILED.", "Esc back.": "Esc retour.", "A add, E edit, D delete. Utilities can load built-ins.": "A ajouter, E modifier, D supprimer. Les utilitaires peuvent charger les types intégrés.", Loading
src/jama/tests.py +21 −0 Changes for src/jama/tests.py: 21 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -379,6 +379,27 @@ class AdminUiAppTestCase(TransactionTestCase): with patch.dict(os.environ, {"DJANGO_ALLOW_ASYNC_UNSAFE": "true"}): asyncio.run(run_app()) def test_ctrl_k_on_user_opens_api_keys(self): user = User.objects.create(username="api-key-user") api_key, _ = APIKey.create_for_key(user, "raw-key") async def run_app(): from jama.adminui.app import JamaAdminApp app = JamaAdminApp() async with app.run_test() as pilot: app.load("users") await pilot.pause() await pilot.press("ctrl+k") await pilot.pause() self.assertEqual(app.mode_name, "user_api_keys") self.assertEqual(app.context["user_id"], user.pk) self.assertEqual(app.selected_object().pk, api_key.pk) with patch.dict(os.environ, {"DJANGO_ALLOW_ASYNC_UNSAFE": "true"}): asyncio.run(run_app()) def test_m_on_project_opens_metadata_sets(self): project = Project.objects.create(label="Metadata project", description="") Loading
src/resources/admin_services/api_keys.py 0 → 100644 +109 −0 Changes for src/resources/admin_services/api_keys.py: 109 added lines, 0 removed lines. Original line number Diff line number Diff line import secrets import string from dataclasses import dataclass from django.contrib.auth.models import User from resources.models import APIKey @dataclass(frozen=True) class APIKeyCreation: api_key: APIKey raw_key: str created: bool generated: bool def generate_api_key(length: int = 32) -> str: alphabet = string.ascii_letters + string.digits return "".join(secrets.choice(alphabet) for _ in range(length)) def list_api_keys(*, user: User | None = None, active_only: bool = False): api_keys = APIKey.objects.select_related("user").order_by("user__username", "id") if user is not None: api_keys = api_keys.filter(user=user) if active_only: api_keys = api_keys.filter(active=True) return api_keys def create_api_key( user: User, raw_key: str | None = None, *, active: bool = True ) -> APIKeyCreation: generated = not raw_key raw_key = raw_key or generate_api_key() api_key, created = APIKey.create_for_key(user=user, key=raw_key) if api_key.active != active: api_key.active = active api_key.save(update_fields=["active"]) return APIKeyCreation(api_key, raw_key, created, generated) def create_api_key_for_username( username: str, raw_key: str | None = None, *, active: bool = True, create_user: bool = False, ) -> APIKeyCreation: user = User.objects.filter(username=username).first() if user is None: if not create_user: raise ValueError( f'User "{username}" does not exist. Use --create-user to create it.' ) user = User.objects.create(username=username) return create_api_key(user, raw_key, active=active) def update_api_key(api_key: APIKey, *, active: bool) -> APIKey: api_key.active = active api_key.save(update_fields=["active"]) return api_key def get_api_key( *, key_id: int | None = None, key_hash: str | None = None, username: str | None = None, raw_key: str | None = None, ) -> APIKey: selectors = sum( selected for selected in ( key_id is not None, key_hash is not None, username is not None or raw_key is not None, ) ) if selectors != 1: raise ValueError( "Delete exactly one key using --id, --hash, or username plus raw key." ) if (username is None) != (raw_key is None): raise ValueError("Deleting by raw key requires both username and key.") filters = {} if key_id is not None: filters["id"] = key_id elif key_hash is not None: filters["key_hash"] = key_hash else: user = User.objects.filter(username=username).first() if user is None: raise ValueError(f'User "{username}" does not exist.') filters.update(user=user, key_hash=APIKey.hash_key(raw_key)) try: return APIKey.objects.select_related("user").get(**filters) except APIKey.DoesNotExist: raise ValueError("API key does not exist.") from None def delete_api_key(api_key: APIKey) -> tuple[int, str]: deleted_id = api_key.pk username = api_key.user.username api_key.delete() return deleted_id, username
src/resources/management/commands/apikey.py +34 −73 Changes for src/resources/management/commands/apikey.py: 34 added lines, 73 removed lines. Original line number Diff line number Diff line import secrets import string from django.contrib.auth.models import User from django.core.management.base import BaseCommand, CommandError from resources.models import APIKey def generate_api_key(length=32): return "".join( secrets.choice(string.ascii_letters + string.digits) for _ in range(length) ) from resources.admin_services import api_keys as api_key_services class Command(BaseCommand): Loading Loading @@ -75,35 +65,34 @@ class Command(BaseCommand): self.handle_delete(options) def handle_add(self, options): username = options["username"] user = User.objects.filter(username=username).first() if user is None: if not options["create_user"]: raise CommandError( f'User "{username}" does not exist. Use --create-user to create it.' ) user = User.objects.create(username=username) raw_key = options["key"] or generate_api_key() api_key, created = APIKey.create_for_key(user=user, key=raw_key) api_key.active = not options["inactive"] api_key.save(update_fields=["active"]) status = "created" if created else "updated" try: result = api_key_services.create_api_key_for_username( options["username"], options["key"], active=not options["inactive"], create_user=options["create_user"], ) except ValueError as error: raise CommandError(str(error)) from error status = "created" if result.created else "updated" self.stdout.write( f"{status}\tid={api_key.pk}\tuser={user.username}\tactive={api_key.active}" f"{status}\tid={result.api_key.pk}\tuser={result.api_key.user.username}" f"\tactive={result.api_key.active}" ) if options["key"] is None: self.stdout.write(f"key\t{raw_key}") if result.generated: self.stdout.write(f"key\t{result.raw_key}") def handle_list(self, options): api_keys = APIKey.objects.select_related("user").order_by( "user__username", "id" ) user = None if options["username"]: api_keys = api_keys.filter(user__username=options["username"]) if options["active_only"]: api_keys = api_keys.filter(active=True) from django.contrib.auth.models import User user = User.objects.filter(username=options["username"]).first() if user is None: return self.stdout.write("id\tuser\tactive\tkey_hash") api_keys = api_key_services.list_api_keys( user=user, active_only=options["active_only"] ) self.stdout.write("id\tuser\tactive\tkey_hash") for api_key in api_keys: Loading @@ -117,42 +106,14 @@ class Command(BaseCommand): username = options["username"] raw_key = options["key"] selectors = sum( 1 for is_selected in ( key_id is not None, key_hash is not None, username is not None or raw_key is not None, ) if is_selected ) if selectors != 1: raise CommandError( "Delete exactly one key using --id, --hash, or username plus raw key." ) if (username is None) != (raw_key is None): raise CommandError("Deleting by raw key requires both username and key.") api_keys = APIKey.objects.select_related("user") if key_id is not None: api_key = self.get_api_key(api_keys, id=key_id) elif key_hash is not None: api_key = self.get_api_key(api_keys, key_hash=key_hash) else: user = User.objects.filter(username=username).first() if user is None: raise CommandError(f'User "{username}" does not exist.') api_key = self.get_api_key( api_keys, user=user, key_hash=APIKey.hash_key(raw_key) ) deleted_id = api_key.pk deleted_user = api_key.user.username api_key.delete() self.stdout.write(f"deleted\tid={deleted_id}\tuser={deleted_user}") def get_api_key(self, api_keys, **filters): try: return api_keys.get(**filters) except APIKey.DoesNotExist: raise CommandError("API key does not exist.") from None api_key = api_key_services.get_api_key( key_id=key_id, key_hash=key_hash, username=username, raw_key=raw_key, ) deleted_id, deleted_user = api_key_services.delete_api_key(api_key) except ValueError as error: raise CommandError(str(error)) from error self.stdout.write(f"deleted\tid={deleted_id}\tuser={deleted_user}")
