refactor(users): move delete_users to admin users blueprint

Move POST /auth/user/delete to POST /auth/system/administration/users/delete,
consistent with the create endpoint already living there.

gn_auth/auth/system/admin/users.py:
- Add imports for sqlite3, reduce, Sequence, AuthorisationError,
  authorised_for2, and system_resource.
- Move __delete_users_individually__, __fetch_non_deletable_users__,
  __non_deletable_with_reason__, and delete_users() from users/views.py.

gn_auth/auth/authorisation/users/views.py:
- Remove the four functions moved to admin/users.py.

The sole known caller (GN2 wqflask/oauth2/users.py) is updated in the
accompanying genenetwork2 patch.

Reviewed-By: Frederick M. Muriithi <fredmanglis@gmail.com>

2 files changed, 172 insertions, 169 deletions

diff --git a/gn_auth/auth/authorisation/users/views.py b/gn_auth/auth/authorisation/users/views.py
index f0b1b0c..f7ac055 100644
--- a/gn_auth/auth/authorisation/users/views.py
+++ b/gn_auth/auth/authorisation/users/views.py
@@ -4,10 +4,10 @@ import logging
 import sqlite3
 import secrets
 import traceback
+from functools import partial
+from typing import Any, Union
 from dataclasses import asdict
 from urllib.parse import urljoin
-from functools import reduce, partial
-from typing import Any, Union, Sequence
 from datetime import datetime, timedelta
 from email.headerregistry import Address
 
@@ -35,9 +35,6 @@ from gn_auth.smtp import send_message, build_email_message
 from gn_auth.auth.requests import request_json
 
 
-from gn_auth.auth.authorisation.resources.system.models import system_resource
-
-from gn_auth.auth.authorisation.resources.checks import authorised_for2
 from gn_auth.auth.authorisation.resources.models import (
     user_resources as _user_resources)
 from gn_auth.auth.authorisation.roles.models import (
@@ -51,7 +48,6 @@ from gn_auth.auth.errors import (
     UsernameError,
     PasswordError,
     ForbiddenAccess,
-    AuthorisationError,
     UserRegistrationError)
 
 
@@ -600,168 +596,6 @@ def change_password(forgot_password_token):
         return change_password_page
 
 
-def __delete_users_individually__(cursor, user_ids, tables):
-    """Recovery function with dismal performance."""
-    _errors = tuple()
-    for _user_id in user_ids:
-        for _table, _col in tables:
-            try:
-                cursor.execute(
-                        f"DELETE FROM {_table} WHERE {_col}=?",
-                        (str(_user_id),))
-            except sqlite3.IntegrityError:
-                _errors = _errors + (
-                    (("user_id", _user_id),
-                     ("reason", f"User has data in table {_table}")),)
-
-    return _errors
-
-
-def __fetch_non_deletable_users__(cursor, ids_and_reasons):
-    """Fetch detail for non-deletable users."""
-    def __merge__(acc, curr):
-        _curr = dict(curr)
-        _this_dict = acc.get(
-            curr["user_id"], {"reasons": tuple()})
-        _this_dict["reasons"] = _this_dict["reasons"] + (_curr["reason"],)
-        return {**acc, curr["user_id"]: _this_dict}
-
-    _reasons_by_id = reduce(__merge__,
-                            (dict(row) for row in ids_and_reasons),
-                            {})
-    _user_ids = tuple(_reasons_by_id.keys())
-    _paramstr = ", ".join(["?"] * len(_user_ids))
-    cursor.execute(f"SELECT * FROM users WHERE user_id IN ({_paramstr})",
-                   _user_ids)
-    return tuple({
-        "user": dict(row),
-        "reasons": _reasons_by_id[row["user_id"]]["reasons"]
-    } for row in cursor.fetchall())
-
-
-def __non_deletable_with_reason__(
-        user_ids: tuple[str, ...],
-        dbrows: Sequence[sqlite3.Row],
-        reason: str
-    ) -> tuple[tuple[tuple[str, str], tuple[str, str]], ...]:
-    """Build a list of 'non-deletable' user objects."""
-    return tuple((("user_id", _uid), ("reason", reason))
-                 for _uid in user_ids
-                 if _uid in tuple(row["user_id"] for row in dbrows))
-
-
-@users.route("/delete", methods=["POST"])
-@require_oauth("profile user role")
-def delete_users():
-    """Delete the specified user."""
-    with (require_oauth.acquire("profile") as _token,
-          db.connection(current_app.config["AUTH_DB"]) as conn,
-          db.cursor(conn) as cursor):
-        if not authorised_for2(conn,
-                               _token.user,
-                               system_resource(conn),
-                               ("system:user:delete-user",)):
-            raise AuthorisationError(
-                "You need the `system:user:delete-user` privilege to delete "
-                "users from the system.")
-
-        _form = request_json()
-        _user_ids = _form.get("user_ids", [])
-        _non_deletable = set()
-        if str(_token.user.user_id) in _user_ids:
-            _non_deletable.add(
-            (("user_id", str(_token.user.user_id),),
-             ("reason", "You are not allowed to delete yourself.")))
-
-        cursor.execute("SELECT user_id FROM group_users")
-        _group_members = tuple(row["user_id"] for row in cursor.fetchall())
-        _non_deletable.update(__non_deletable_with_reason__(
-            _user_ids,
-            cursor.fetchall(),
-            "User is member of a user group."))
-
-        cursor.execute("SELECT user_id FROM oauth2_clients;")
-        _non_deletable.update(__non_deletable_with_reason__(
-            _user_ids,
-            cursor.fetchall(),
-            "User is registered owner of an OAuth client."))
-
-        _important_roles = (
-            "group-leader",
-            "resource-owner",
-            "system-administrator",
-            "inbredset-group-owner")
-        _paramstr = ",".join(["?"] * len(_important_roles))
-        cursor.execute(
-            "SELECT DISTINCT user_roles.user_id FROM user_roles "
-            "INNER JOIN roles ON user_roles.role_id=roles.role_id "
-            f"WHERE roles.role_name IN ({_paramstr})",
-            _important_roles)
-        _non_deletable.update(__non_deletable_with_reason__(
-            _user_ids,
-            cursor.fetchall(),
-            f"User holds on of the following roles: {_important_roles}"))
-
-        _delete = tuple(uid for uid in _user_ids if uid not in
-                        (dict(row)["user_id"] for row in _non_deletable))
-        _paramstr = ", ".join(["?"] * len(_delete))
-        if len(_delete) > 0:
-            _dependent_tables = (
-                ("authorisation_code", "user_id"),
-                ("forgot_password_tokens", "user_id"),
-                ("group_join_requests", "requester_id"),
-                ("jwt_refresh_tokens", "user_id"),
-                ("oauth2_tokens", "user_id"),
-                ("user_credentials", "user_id"),
-                ("user_roles", "user_id"),
-                ("user_verification_codes", "user_id"))
-            try:
-                for _table, _col in _dependent_tables:
-                    cursor.execute(
-                        f"DELETE FROM {_table} WHERE {_col} IN ({_paramstr})",
-                        _delete)
-            except sqlite3.IntegrityError:
-                _non_deletable.update(__delete_users_individually__(
-                    cursor, _delete, _dependent_tables))
-
-            _not_deleted = __fetch_non_deletable_users__(
-                cursor, _non_deletable)
-            _delete = tuple(# rebuild with those that failed.
-                _user_id for _user_id in _delete if _user_id not in
-                tuple(row["user"]["user_id"] for row in _not_deleted))
-            _paramstr = ", ".join(["?"] * len(_delete))
-            cursor.execute(
-                f"DELETE FROM users WHERE user_id IN ({_paramstr})",
-                _delete)
-            _deleted_rows = cursor.rowcount
-            return jsonify({
-                "total-requested": len(_user_ids),
-                "total-deleted": _deleted_rows,
-                "not-deleted": _not_deleted,
-                "deleted": _deleted_rows,
-                "message": (
-                    f"Successfully deleted {_deleted_rows} users." +
-                    (" Some users could not be deleted."
-                     if len(_user_ids) - _deleted_rows > 0
-                     else ""))
-            })
-
-        _not_deleted = __fetch_non_deletable_users__(cursor, _non_deletable)
-
-    return jsonify({
-        "total-requested": len(_user_ids),
-        "total-deleted": 0,
-        "not-deleted": _not_deleted,
-        "deleted": 0,
-        "error": "Zero users were deleted",
-        "error_description": (
-            "No users were selected for deletion."
-            if len(_user_ids) == 0
-            else ("The selected users are system administrators, group "
-                  "members, or resource owners."))
-    }), 400
-
-
 @users.route("/<uuid:user_id>/roles/assign", methods=["POST"])
 def assign_user_role(user_id: uuid.UUID) -> Response:
     """Assign a role to a user on a given resource."""
diff --git a/gn_auth/auth/system/admin/users.py b/gn_auth/auth/system/admin/users.py
index f26518b..63648d8 100644
--- a/gn_auth/auth/system/admin/users.py
+++ b/gn_auth/auth/system/admin/users.py
@@ -1,4 +1,8 @@
 """Administrative endpoints for user management."""
+import sqlite3
+from functools import reduce
+from typing import Sequence
+
 from flask import jsonify, Response, Blueprint, make_response, current_app as app
 
 from gn_libs import sqlite3 as db
@@ -11,10 +15,14 @@ from gn_auth.auth.errors import (
     PasswordError,
     UsernameError,
     ForbiddenAccess,
+    AuthorisationError,
     UserRegistrationError)
 from gn_auth.auth.requests import request_json
 from gn_auth.auth.authentication.oauth2.resource_server import require_oauth
-from gn_auth.auth.authorisation.resources.system.models import user_roles_on_system
+from gn_auth.auth.authorisation.resources.checks import authorised_for2
+from gn_auth.auth.authorisation.resources.system.models import (
+    system_resource,
+    user_roles_on_system)
 from gn_auth.auth.authorisation.users.admin.models import create_verified_user
 from gn_auth.auth.authorisation.users.views import (
     validate_password,
@@ -65,3 +73,164 @@ def create_user() -> Response:
         "email": user.email,
         "name": user.name
     }), 201)
+
+
+def __delete_users_individually__(cursor, user_ids, tables):
+    """Recovery function with dismal performance."""
+    _errors = tuple()
+    for _user_id in user_ids:
+        for _table, _col in tables:
+            try:
+                cursor.execute(
+                        f"DELETE FROM {_table} WHERE {_col}=?",
+                        (str(_user_id),))
+            except sqlite3.IntegrityError:
+                _errors = _errors + (
+                    (("user_id", _user_id),
+                     ("reason", f"User has data in table {_table}")),)
+
+    return _errors
+
+
+def __fetch_non_deletable_users__(cursor, ids_and_reasons):
+    """Fetch detail for non-deletable users."""
+    def __merge__(acc, curr):
+        _curr = dict(curr)
+        _this_dict = acc.get(
+            curr["user_id"], {"reasons": tuple()})
+        _this_dict["reasons"] = _this_dict["reasons"] + (_curr["reason"],)
+        return {**acc, curr["user_id"]: _this_dict}
+
+    _reasons_by_id = reduce(__merge__,
+                            (dict(row) for row in ids_and_reasons),
+                            {})
+    _user_ids = tuple(_reasons_by_id.keys())
+    _paramstr = ", ".join(["?"] * len(_user_ids))
+    cursor.execute(f"SELECT * FROM users WHERE user_id IN ({_paramstr})",
+                   _user_ids)
+    return tuple({
+        "user": dict(row),
+        "reasons": _reasons_by_id[row["user_id"]]["reasons"]
+    } for row in cursor.fetchall())
+
+
+def __non_deletable_with_reason__(
+        user_ids: tuple[str, ...],
+        dbrows: Sequence[sqlite3.Row],
+        reason: str
+    ) -> tuple[tuple[tuple[str, str], tuple[str, str]], ...]:
+    """Build a list of 'non-deletable' user objects."""
+    return tuple((("user_id", _uid), ("reason", reason))
+                 for _uid in user_ids
+                 if _uid in tuple(row["user_id"] for row in dbrows))
+
+
+@users.route("/delete", methods=["POST"])
+def delete_users() -> Response:
+    """Delete the specified users. Requires system:user:delete-user privilege."""
+    with (require_oauth.acquire("profile user role") as _token,
+          db.connection(app.config["AUTH_DB"]) as conn,
+          db.cursor(conn) as cursor):
+        if not authorised_for2(conn,
+                               _token.user,
+                               system_resource(conn),
+                               ("system:user:delete-user",)):
+            raise AuthorisationError(
+                "You need the `system:user:delete-user` privilege to delete "
+                "users from the system.")
+
+        _form = request_json()
+        _user_ids = _form.get("user_ids", [])
+        _non_deletable = set()
+        if str(_token.user.user_id) in _user_ids:
+            _non_deletable.add(
+            (("user_id", str(_token.user.user_id),),
+             ("reason", "You are not allowed to delete yourself.")))
+
+        cursor.execute("SELECT user_id FROM group_users")
+        _group_members = tuple(row["user_id"] for row in cursor.fetchall())
+        _non_deletable.update(__non_deletable_with_reason__(
+            _user_ids,
+            cursor.fetchall(),
+            "User is member of a user group."))
+
+        cursor.execute("SELECT user_id FROM oauth2_clients;")
+        _non_deletable.update(__non_deletable_with_reason__(
+            _user_ids,
+            cursor.fetchall(),
+            "User is registered owner of an OAuth client."))
+
+        _important_roles = (
+            "group-leader",
+            "resource-owner",
+            "system-administrator",
+            "inbredset-group-owner")
+        _paramstr = ",".join(["?"] * len(_important_roles))
+        cursor.execute(
+            "SELECT DISTINCT user_roles.user_id FROM user_roles "
+            "INNER JOIN roles ON user_roles.role_id=roles.role_id "
+            f"WHERE roles.role_name IN ({_paramstr})",
+            _important_roles)
+        _non_deletable.update(__non_deletable_with_reason__(
+            _user_ids,
+            cursor.fetchall(),
+            f"User holds on of the following roles: {_important_roles}"))
+
+        _delete = tuple(uid for uid in _user_ids if uid not in
+                        (dict(row)["user_id"] for row in _non_deletable))
+        _paramstr = ", ".join(["?"] * len(_delete))
+        if len(_delete) > 0:
+            _dependent_tables = (
+                ("authorisation_code", "user_id"),
+                ("forgot_password_tokens", "user_id"),
+                ("group_join_requests", "requester_id"),
+                ("jwt_refresh_tokens", "user_id"),
+                ("oauth2_tokens", "user_id"),
+                ("user_credentials", "user_id"),
+                ("user_roles", "user_id"),
+                ("user_verification_codes", "user_id"))
+            try:
+                for _table, _col in _dependent_tables:
+                    cursor.execute(
+                        f"DELETE FROM {_table} WHERE {_col} IN ({_paramstr})",
+                        _delete)
+            except sqlite3.IntegrityError:
+                _non_deletable.update(__delete_users_individually__(
+                    cursor, _delete, _dependent_tables))
+
+            _not_deleted = __fetch_non_deletable_users__(
+                cursor, _non_deletable)
+            _delete = tuple(# rebuild with those that failed.
+                _user_id for _user_id in _delete if _user_id not in
+                tuple(row["user"]["user_id"] for row in _not_deleted))
+            _paramstr = ", ".join(["?"] * len(_delete))
+            cursor.execute(
+                f"DELETE FROM users WHERE user_id IN ({_paramstr})",
+                _delete)
+            _deleted_rows = cursor.rowcount
+            return jsonify({
+                "total-requested": len(_user_ids),
+                "total-deleted": _deleted_rows,
+                "not-deleted": _not_deleted,
+                "deleted": _deleted_rows,
+                "message": (
+                    f"Successfully deleted {_deleted_rows} users." +
+                    (" Some users could not be deleted."
+                     if len(_user_ids) - _deleted_rows > 0
+                     else ""))
+            })
+
+        _not_deleted = __fetch_non_deletable_users__(cursor, _non_deletable)
+
+    return make_response(jsonify({
+        "total-requested": len(_user_ids),
+        "total-deleted": 0,
+        "not-deleted": _not_deleted,
+        "deleted": 0,
+        "error": "Zero users were deleted",
+        "error_description": (
+            "No users were selected for deletion."
+            if len(_user_ids) == 0
+            else ("The selected users are system administrators, group "
+                  "members, or resource owners."))
+    }), 400)