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>

1 files changed, 2 insertions, 168 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."""