feat: feat: Add remove all users blong to an admin to tui and cil

Author: ImMohammad20000

cli/admin.py

@@ -117,17 +117,58 @@ async def delete_admin(self, db, username: str):
 
         # Check if admin exists
         admins = await admin_op.get_admins(db)
-        if not any(admin.username == username for admin in admins):
+        target_admin = next((admin for admin in admins if admin.username == username), None)
+        if not target_admin:
             self.console.print(f"[red]Admin '{username}' not found[/red]")
             return
 
+        user_count = len(target_admin.users or [])
+
         if typer.confirm(f"Are you sure you want to delete admin '{username}'?"):
+            if user_count > 0:
+                message = (
+                    f"Admin '{username}' owns {user_count} users. Delete all of their users before removing the admin?"
+                )
+                delete_users = typer.confirm(message, default=False)
+                if delete_users:
+                    try:
+                        await admin_op.remove_all_users(db, username, SYSTEM_ADMIN)
+                        self.console.print(f"[green]Deleted {user_count} users belonging to admin '{username}'[/green]")
+                    except Exception as e:
+                        self.console.print(f"[red]Error deleting users: {e}[/red]")
+                        return
+
             try:
                 await admin_op.remove_admin(db, username, SYSTEM_ADMIN)
                 self.console.print(f"[green]Admin '{username}' deleted successfully[/green]")
             except Exception as e:
                 self.console.print(f"[red]Error deleting admin: {e}[/red]")
 
+    async def delete_admin_users(self, db, username: str):
+        """Delete all users belonging to an admin."""
+        admin_op = get_admin_operation()
+
+        admins = await admin_op.get_admins(db)
+        target_admin = next((admin for admin in admins if admin.username == username), None)
+        if not target_admin:
+            self.console.print(f"[red]Admin '{username}' not found[/red]")
+            return
+
+        if not typer.confirm(
+            f"Delete all users belonging to admin '{username}'? This action cannot be undone.", default=False
+        ):
+            self.console.print("[yellow]Operation cancelled[/yellow]")
+            return
+
+        try:
+            deleted = await admin_op.remove_all_users(db, username, SYSTEM_ADMIN)
+            if deleted == 0:
+                self.console.print(f"[yellow]Admin '{username}' has no users to delete[/yellow]")
+            else:
+                self.console.print(f"[green]Deleted {deleted} users belonging to admin '{username}'[/green]")
+        except Exception as e:
+            self.console.print(f"[red]Error deleting users: {e}[/red]")
+
     async def modify_admin(self, db, username: str, disable: bool):
         """Modify an admin account."""
         admin_op = get_admin_operation()
@@ -193,17 +234,15 @@ async def modify_admin(self, db, username: str, disable: bool):
             self.console.print("\n[cyan]Notification Preferences:[/cyan]")
             enable_notifications = typer.confirm(
                 "Enable user notifications for this admin?",
-                default=any(
-                    [
-                        current_admin.notification_enable["create"],
-                        current_admin.notification_enable["modify"],
-                        current_admin.notification_enable["delete"],
-                        current_admin.notification_enable["status_change"],
-                        current_admin.notification_enable["reset_data_usage"],
-                        current_admin.notification_enable["data_reset_by_next"],
-                        current_admin.notification_enable["subscription_revoked"],
-                    ]
-                ),
+                default=any([
+                    current_admin.notification_enable["create"],
+                    current_admin.notification_enable["modify"],
+                    current_admin.notification_enable["delete"],
+                    current_admin.notification_enable["status_change"],
+                    current_admin.notification_enable["reset_data_usage"],
+                    current_admin.notification_enable["data_reset_by_next"],
+                    current_admin.notification_enable["subscription_revoked"],
+                ]),
             )
 
             if enable_notifications:
@@ -318,6 +357,15 @@ async def delete_admin(username: str):
             console.print(f"[red]Error: {e}[/red]")
 
 
+async def delete_admin_users(username: str):
+    """Delete all users belonging to an admin."""
+    async with GetDB() as db:
+        try:
+            await admin_cli.delete_admin_users(db, username)
+        except Exception as e:
+            console.print(f"[red]Error: {e}[/red]")
+
+
 async def modify_admin(
     username: str,
     disable: Annotated[bool, typer.Option(..., "--disable", help="Disable or enable the admin account.")] = None,

cli/main.py

@@ -11,7 +11,7 @@
 import typer
 
 from cli import console
-from cli.admin import create_admin, delete_admin, list_admins, modify_admin, reset_admin_usage
+from cli.admin import create_admin, delete_admin, delete_admin_users, list_admins, modify_admin, reset_admin_usage
 from cli.system import show_status
 
 # Initialize Typer app
@@ -37,18 +37,23 @@ def admins(
     create: Optional[str] = typer.Option(None, "--create", "-c", help="Create new admin"),
     sudo: bool = typer.Option(False, "--sudo", "-s", help="Create a sudo admin."),
     delete: Optional[str] = typer.Option(None, "--delete", "-d", help="Delete admin"),
+    delete_users: Optional[str] = typer.Option(
+        None, "--delete-users", "-u", help="Delete all users belonging to an admin"
+    ),
     modify: Optional[str] = typer.Option(None, "--modify", "-m", help="Modify admin"),
     disable: Optional[bool] = typer.Option(None, "--disable", help="Disable or enable the admin account."),
     reset_usage: Optional[str] = typer.Option(None, "--reset-usage", "-r", help="Reset admin usage"),
 ):
     """List & manage admin accounts."""
 
-    if list or not any([create, delete, modify, reset_usage]):
+    if list or not any([create, delete, delete_users, modify, reset_usage]):
         asyncio.run(list_admins())
     elif create:
         asyncio.run(create_admin(create, sudo))
     elif delete:
         asyncio.run(delete_admin(delete))
+    elif delete_users:
+        asyncio.run(delete_admin_users(delete_users))
     elif modify:
         asyncio.run(modify_admin(modify, disable))
     elif reset_usage:

tui/admin.py

@@ -27,22 +27,40 @@
 
 class AdminDelete(BaseModal):
     def __init__(
-        self, db: AsyncSession, operation: AdminOperation, username: str, on_close: callable, *args, **kwargs
+        self,
+        db: AsyncSession,
+        operation: AdminOperation,
+        username: str,
+        on_close: callable,
+        user_count: int = 0,
+        *args,
+        **kwargs,
     ) -> None:
         super().__init__(*args, **kwargs)
         self.db = db
         self.operation = operation
         self.username = username
         self.on_close = on_close
+        self.user_count = user_count
 
     async def on_mount(self) -> None:
         """Ensure the first button is focused."""
-        yes_button = self.query_one("#no")
-        self.set_focus(yes_button)
+        focus_target = "#delete-users" if self.user_count > 0 else "#no"
+        self.set_focus(self.query_one(focus_target))
 
     def compose(self) -> ComposeResult:
         with Container(classes="modal-box-delete"):
-            yield Static("Are you sure about deleting this admin?", classes="title")
+            yield Static(f"Delete admin '{self.username}'?", classes="title")
+            if self.user_count > 0:
+                yield Static(
+                    f"This admin has {self.user_count} users.\nYou must delete them to remove the admin.",
+                    classes="subtitle",
+                )
+                yield Horizontal(
+                    Static("Delete all users:", classes="label"),
+                    Switch(animate=False, id="delete-users"),
+                    classes="switch-container",
+                )
             yield Horizontal(
                 Button("Yes", id="yes", variant="success"),
                 Button("No", id="no", variant="error"),
@@ -52,13 +70,69 @@ def compose(self) -> ComposeResult:
     async def on_button_pressed(self, event: Button.Pressed) -> None:
         if event.button.id == "yes":
             try:
+                if self.user_count > 0:
+                    delete_users = self.query_one("#delete-users").value
+                    if delete_users:
+                        await self.operation.remove_all_users(self.db, self.username, SYSTEM_ADMIN)
+                        self.notify("Admin users deleted successfully", severity="success", title="Success")
                 await self.operation.remove_admin(self.db, self.username, SYSTEM_ADMIN)
                 self.on_close()
             except ValueError as e:
                 self.notify(str(e), severity="error", title="Error")
         await self.key_escape()
 
 
+class AdminDeleteUsers(BaseModal):
+    def __init__(
+        self,
+        db: AsyncSession,
+        operation: AdminOperation,
+        username: str,
+        on_close: callable,
+        user_count: int = 0,
+        *args,
+        **kwargs,
+    ) -> None:
+        super().__init__(*args, **kwargs)
+        self.db = db
+        self.operation = operation
+        self.username = username
+        self.on_close = on_close
+        self.user_count = user_count
+
+    async def on_mount(self) -> None:
+        confirm_button = self.query_one("#cancel")
+        self.set_focus(confirm_button)
+
+    def compose(self) -> ComposeResult:
+        with Container(classes="modal-box-delete"):
+            yield Static(
+                f"Delete all users belonging to admin '{self.username}'?"
+                f"\nFound {self.user_count} user(s). This action cannot be undone.",
+                classes="title",
+            )
+            yield Horizontal(
+                Button("Delete Users", id="delete", variant="warning"),
+                Button("Cancel", id="cancel", variant="error"),
+                classes="button-container",
+            )
+
+    async def on_button_pressed(self, event: Button.Pressed) -> None:
+        if event.button.id == "delete":
+            try:
+                deleted = await self.operation.remove_all_users(self.db, self.username, SYSTEM_ADMIN)
+                if deleted == 0:
+                    self.notify("No users were deleted (none found)", severity="warning", title="Info")
+                else:
+                    self.notify(
+                        f"{deleted} users deleted for admin '{self.username}'", severity="success", title="Success"
+                    )
+                self.on_close()
+            except ValueError as e:
+                self.notify(str(e), severity="error", title="Error")
+        await self.key_escape()
+
+
 class AdminResetUsage(BaseModal):
     def __init__(
         self, db: AsyncSession, operation: AdminOperation, username: str, on_close: callable, *args, **kwargs
@@ -396,17 +470,15 @@ async def on_mount(self) -> None:
 
         # Load existing notification preferences (notification_enable is a dict from SQLAlchemy)
         notif = self.admin.notification_enable or {}
-        master_on = any(
-            [
-                notif.get("create", False),
-                notif.get("modify", False),
-                notif.get("delete", False),
-                notif.get("status_change", False),
-                notif.get("reset_data_usage", False),
-                notif.get("data_reset_by_next", False),
-                notif.get("subscription_revoked", False),
-            ]
-        )
+        master_on = any([
+            notif.get("create", False),
+            notif.get("modify", False),
+            notif.get("delete", False),
+            notif.get("status_change", False),
+            notif.get("reset_data_usage", False),
+            notif.get("data_reset_by_next", False),
+            notif.get("subscription_revoked", False),
+        ])
 
         self.query_one("#notif_master").value = master_on
         self.query_one("#notif_create").value = notif.get("create", False)
@@ -549,6 +621,7 @@ def __init__(self, *args, **kwargs) -> None:
         ("m", "modify_admin", "Modify admin"),
         ("r", "reset_admin_usage", "Reset admin usage"),
         ("d", "delete_admin", "Delete admin"),
+        ("u", "delete_admin_users", "Delete admin users"),
         ("i", "import_from_env", "Import from env"),
         ("p", "previous_page", "Previous page"),
         ("n", "next_page", "Next page"),
@@ -648,7 +721,20 @@ def selected_admin(self):
     async def action_delete_admin(self):
         if not self.table.columns:
             return
-        self.app.push_screen(AdminDelete(self.db, self.admin_operator, self.selected_admin, self._refresh_table))
+        admin = await self.admin_operator.get_validated_admin(self.db, username=self.selected_admin)
+        user_count = len(admin.users or [])
+        self.app.push_screen(
+            AdminDelete(self.db, self.admin_operator, self.selected_admin, self._refresh_table, user_count)
+        )
+
+    async def action_delete_admin_users(self):
+        if not self.table.columns:
+            return
+        admin = await self.admin_operator.get_validated_admin(self.db, username=self.selected_admin)
+        user_count = len(admin.users or [])
+        self.app.push_screen(
+            AdminDeleteUsers(self.db, self.admin_operator, self.selected_admin, self._refresh_table, user_count)
+        )
 
     def _refresh_table(self):
         self.run_worker(self.admins_list)