auth: redis data: migrate data in redis

Implement the code to migrate the data from redis to SQLite.

5 files changed, 237 insertions, 107 deletions

diff --git a/gn3/auth/authorisation/data/views.py b/gn3/auth/authorisation/data/views.py
index 89898c6..1343f47 100644
--- a/gn3/auth/authorisation/data/views.py
+++ b/gn3/auth/authorisation/data/views.py
@@ -1,20 +1,42 @@
 """Handle data endpoints."""
+import os
 import uuid
 import json
+import datetime
+from typing import Sequence
+from functools import reduce
 
+import redis
 from email_validator import validate_email, EmailNotValidError
 from authlib.integrations.flask_oauth2.errors import _HTTPException
 from flask import request, jsonify, Response, Blueprint, current_app as app
 
+import gn3.db_utils as gn3db
 from gn3.db.traits import build_trait_name
 
 from gn3.auth import db
-from gn3.auth.authorisation.users.views import validate_password
+from gn3.auth.dictify import dictify
+
+from gn3.auth.authorisation.errors import NotFoundError
+
+from gn3.auth.authorisation.users.views import (
+    validate_password, validate_username)
+
+from gn3.auth.authorisation.roles.models import(
+    revoke_user_role_by_name, assign_user_role_by_name)
+
+from gn3.auth.authorisation.groups.data import retrieve_ungrouped_data
+from gn3.auth.authorisation.groups.models import (
+    Group, user_group, add_user_to_group)
+
 from gn3.auth.authorisation.resources.checks import authorised_for
-from gn3.auth.authorisation.errors import ForbiddenAccess, AuthorisationError
 from gn3.auth.authorisation.resources.models import (
     user_resources, public_resources, attach_resources_data)
 
+from gn3.auth.authorisation.errors import ForbiddenAccess, AuthorisationError
+
+
+from gn3.auth.authentication.users import User, user_by_id, set_user_password
 from gn3.auth.authentication.oauth2.resource_server import require_oauth
 
 data = Blueprint("data", __name__)
@@ -88,6 +110,101 @@ def authorisation() -> Response:
             (build_trait_name(trait_fullname)
              for trait_fullname in traits_names)))
 
+def migrate_user(conn: db.DbConnection, user_id: uuid.UUID, email: str,
+                 username: str, password: str) -> User:
+    """Migrate the user, if not already migrated."""
+    try:
+        return user_by_id(conn, user_id)
+    except NotFoundError as _nfe:
+        user = User(user_id, email, username)
+        with db.cursor(conn) as cursor:
+            cursor.execute(
+                "INSERT INTO users(user_id, email, name) "
+                "VALUES (?, ?, ?)",
+                (str(user.user_id), user.email, user.name))
+            set_user_password(cursor, user, password)
+            return user
+
+def migrate_user_group(conn: db.DbConnection, user: User) -> Group:
+    """Create a group for the user if they don't already have a group."""
+    group = user_group(conn, user).maybe(# type: ignore[misc]
+        False, lambda grp: grp) # type: ignore[arg-type]
+    if not bool(group):
+        group = Group(uuid.UUID(), f"{user.name}'s Group", {
+            "created": datetime.datetime.now().isoformat(),
+            "notes": "Imported from redis"
+        })
+        with db.cursor(conn) as cursor:
+            cursor.execute(
+                "INSERT INTO groups(group_id, group_name, group_metadata) "
+                "VALUES(?, ?, ?)",
+                (str(group.group_id), group.group_name, json.dumps(
+                    group.group_metadata)))
+            add_user_to_group(cursor, group, user)
+            revoke_user_role_by_name(cursor, user, "group-creator")
+            assign_user_role_by_name(cursor, user, "group-leader")
+
+    return group
+
+def __redis_datasets_by_type__(acc, item):
+    if item["type"] == "dataset-probeset":
+        return (acc[0] + (item["name"],), acc[1], acc[2])
+    if item["type"] == "dataset-geno":
+        return (acc[0], acc[1] + (item["name"],), acc[2])
+    if item["type"] == "dataset-publish":
+        return (acc[0], acc[1], acc[2] + (item["name"],))
+    return acc
+
+def __unmigrated_data__(ungrouped, redis_datasets):
+    return (dataset for dataset in ungrouped
+            if dataset["Name"] in redis_datasets)
+
+def __parametrise__(group: Group, datasets: Sequence[dict],
+                    dataset_type: str) -> tuple[dict[str, str], ...]:
+    return tuple(
+        {
+            "group_id": str(group.group_id),
+            "dataset_type": dataset_type,
+            "dataset_or_trait_id": dataset["Id"],
+            "dataset_name": dataset["Name"],
+            "dataset_fullname": dataset["FullName"],
+            "accession_id": dataset["accession_id"]
+        } for dataset in datasets)
+
+def migrate_data(
+        authconn: db.DbConnection, gn3conn: gn3db.Connection,
+        rconn: redis.Redis, user: User,
+        group: Group) -> tuple[dict[str, str], ...]:
+    """Migrate data attached to the user to the user's group."""
+    redis_mrna, redis_geno, redis_pheno = reduce(# type: ignore[var-annotated]
+        __redis_datasets_by_type__,
+        (dataset for dataset in
+         (dataset for _key,dataset in {
+             key: json.loads(val)
+            for key,val in rconn.hgetall("resources").items()
+         }.items())
+         if dataset["owner_id"] == str(user.user_id)),
+        (tuple(), tuple(), tuple()))
+    mrna_datasets = __unmigrated_data__(
+        retrieve_ungrouped_data(authconn, gn3conn, "mrna"), redis_mrna)
+    geno_datasets = __unmigrated_data__(
+        retrieve_ungrouped_data(authconn, gn3conn, "genotype"), redis_geno)
+    pheno_datasets = __unmigrated_data__(
+        retrieve_ungrouped_data(authconn, gn3conn, "phenotype"), redis_pheno)
+    params = (
+        __parametrise__(group, mrna_datasets, "mRNA") +
+        __parametrise__(group, geno_datasets, "Genotype") +
+        __parametrise__(group, pheno_datasets, "Phenotype"))
+    if len(params) > 0:
+        with db.cursor(authconn) as cursor:
+            cursor.executemany(
+                "INSERT INTO linked_group_data VALUES"
+                "(:group_id, :dataset_type, :dataset_or_trait_id, "
+                ":dataset_name, :dataset_fullname, :accession_id)",
+                params)
+
+    return params
+
 @data.route("/user/migrate", methods=["POST"])
 @require_oauth("migrate-data")
 def migrate_user_data():
@@ -98,35 +215,47 @@ def migrate_user_data():
     This is a temporary endpoint and should be removed after all the data has
     been migrated.
     """
-    authorised_clients = app.config.get(
-        "OAUTH2_CLIENTS_WITH_DATA_MIGRATION_PRIVILEGE", [])
-    with require_oauth.acquire("migrate-data") as the_token:
-        if the_token.client.client_id in authorised_clients:
-            try:
-                _user_id = uuid.UUID(request.form.get("user_id", ""))
-                _email = validate_email(request.form.get("email", ""))
-                _password = validate_password(
-                    request.form.get("password", ""),
-                    request.form.get("confirm_password", ""))
-                ## TODO: Save the user: possible exception for duplicate emails
-                ##       Create group from user's name
-                ##       Filter all resources from redis owned by this user
-                ##         resources = {key: json.loads(val)
-                ##                      for key,val
-                ##                      in rconn.hgetall("resources").items()}
-                ##         filtered = dict((
-                ##             (key,val) for key,val
-                ##             in resources.items()
-                ##             if uuid.UUID(val.get("owner_id")) == user_id))
-                ##       Check that no resource is owned by existing user, use
-                ##         'name' and 'type' fields to check in
-                ##         `linked_group_data` table
-                ##       Link remaining data to the new group
-                ##       Delete user from redis
-                return "WOULD TRIGGER DATA MIGRATION ..."
-            except EmailNotValidError as enve:
-                raise AuthorisationError(f"Email Error: {str(enve)}") from enve
-            except ValueError as verr:
-                raise AuthorisationError(verr.args[0]) from verr
-
-        raise ForbiddenAccess("You cannot access this endpoint.")
+    db_uri = app.config.get("AUTH_DB").strip()
+    if bool(db_uri) and os.path.exists(db_uri):
+        authorised_clients = app.config.get(
+            "OAUTH2_CLIENTS_WITH_DATA_MIGRATION_PRIVILEGE", [])
+        with require_oauth.acquire("migrate-data") as the_token:
+            if the_token.client.client_id in authorised_clients:
+                try:
+                    user_id = uuid.UUID(request.form.get("user_id", ""))
+                    email = validate_email(request.form.get("email", ""))
+                    username = validate_username(
+                        request.form.get("username", ""))
+                    password = validate_password(
+                        request.form.get("password", ""),
+                        request.form.get("confirm_password", ""))
+
+                    with (db.connection(db_uri) as authconn,
+                          redis.Redis(decode_responses=True) as rconn,
+                          gn3db.database_connection() as gn3conn):
+                        user = migrate_user(
+                            authconn, user_id, email["email"], username,
+                            password)
+                        group = migrate_user_group(authconn, user)
+                        user_resource_data = migrate_data(
+                            authconn, gn3conn, rconn, user, group)
+                        ## TODO: Maybe delete user from redis...
+                        return jsonify({
+                            "description": (
+                                f"Migrated {len(user_resource_data)} resource data "
+                                "items."),
+                            "user": dictify(user),
+                            "group": dictify(group)
+                        })
+                except EmailNotValidError as enve:
+                    raise AuthorisationError(f"Email Error: {str(enve)}") from enve
+                except ValueError as verr:
+                    raise AuthorisationError(verr.args[0]) from verr
+
+            raise ForbiddenAccess("You cannot access this endpoint.")
+
+    return jsonify({
+        "error": "Unavailable",
+        "error_description": (
+            "The data migration service is currently unavailable.")
+    }), 503
diff --git a/gn3/auth/authorisation/groups/models.py b/gn3/auth/authorisation/groups/models.py
index 5a58322..bbe4ad6 100644
--- a/gn3/auth/authorisation/groups/models.py
+++ b/gn3/auth/authorisation/groups/models.py
@@ -142,30 +142,31 @@ def authenticated_user_group(conn) -> Maybe:
 
     return Nothing
 
-def user_group(cursor: db.DbCursor, user: User) -> Maybe[Group]:
+def user_group(conn: db.DbConnection, user: User) -> Maybe[Group]:
     """Returns the given user's group"""
-    cursor.execute(
-        ("SELECT groups.group_id, groups.group_name, groups.group_metadata "
-         "FROM group_users "
-         "INNER JOIN groups ON group_users.group_id=groups.group_id "
-         "WHERE group_users.user_id = ?"),
-        (str(user.user_id),))
-    groups = tuple(
-        Group(UUID(row[0]), row[1], json.loads(row[2] or "{}"))
-        for row in cursor.fetchall())
+    with db.cursor(conn) as cursor:
+        cursor.execute(
+            ("SELECT groups.group_id, groups.group_name, groups.group_metadata "
+             "FROM group_users "
+             "INNER JOIN groups ON group_users.group_id=groups.group_id "
+             "WHERE group_users.user_id = ?"),
+            (str(user.user_id),))
+        groups = tuple(
+            Group(UUID(row[0]), row[1], json.loads(row[2] or "{}"))
+            for row in cursor.fetchall())
 
-    if len(groups) > 1:
-        raise MembershipError(user, groups)
+        if len(groups) > 1:
+            raise MembershipError(user, groups)
 
-    if len(groups) == 1:
-        return Just(groups[0])
+        if len(groups) == 1:
+            return Just(groups[0])
 
     return Nothing
 
-def is_group_leader(cursor: db.DbCursor, user: User, group: Group):
+def is_group_leader(conn: db.DbConnection, user: User, group: Group) -> bool:
     """Check whether the given `user` is the leader of `group`."""
 
-    ugroup = user_group(cursor, user).maybe(
+    ugroup = user_group(conn, user).maybe(
         False, lambda val: val) # type: ignore[arg-type, misc]
     if not group:
         # User cannot be a group leader if not a member of ANY group
@@ -175,13 +176,14 @@ def is_group_leader(cursor: db.DbCursor, user: User, group: Group):
         # User cannot be a group leader if not a member of THIS group
         return False
 
-    cursor.execute(
-        ("SELECT roles.role_name FROM user_roles LEFT JOIN roles "
-         "ON user_roles.role_id = roles.role_id WHERE user_id = ?"),
-        (str(user.user_id),))
-    role_names = tuple(row[0] for row in cursor.fetchall())
+    with db.cursor(conn) as cursor:
+        cursor.execute(
+            ("SELECT roles.role_name FROM user_roles LEFT JOIN roles "
+             "ON user_roles.role_id = roles.role_id WHERE user_id = ?"),
+            (str(user.user_id),))
+        role_names = tuple(row[0] for row in cursor.fetchall())
 
-    return "group-leader" in role_names
+        return "group-leader" in role_names
 
 def all_groups(conn: db.DbConnection) -> Maybe[Sequence[Group]]:
     """Retrieve all existing groups"""
@@ -258,8 +260,8 @@ def group_by_id(conn: db.DbConnection, group_id: UUID) -> Group:
 def join_requests(conn: db.DbConnection, user: User):
     """List all the join requests for the user's group."""
     with db.cursor(conn) as cursor:
-        group = user_group(cursor, user).maybe(DUMMY_GROUP, lambda grp: grp)# type: ignore[misc]
-        if group != DUMMY_GROUP and is_group_leader(cursor, user, group):
+        group = user_group(conn, user).maybe(DUMMY_GROUP, lambda grp: grp)# type: ignore[misc]
+        if group != DUMMY_GROUP and is_group_leader(conn, user, group):
             cursor.execute(
                 "SELECT gjr.*, u.email, u.name FROM group_join_requests AS gjr "
                 "INNER JOIN users AS u ON gjr.requester_id=u.user_id "
@@ -280,7 +282,7 @@ def accept_reject_join_request(
     """Accept/Reject a join request."""
     assert status in ("ACCEPTED", "REJECTED"), f"Invalid status '{status}'."
     with db.cursor(conn) as cursor:
-        group = user_group(cursor, user).maybe(DUMMY_GROUP, lambda grp: grp) # type: ignore[misc]
+        group = user_group(conn, user).maybe(DUMMY_GROUP, lambda grp: grp) # type: ignore[misc]
         cursor.execute("SELECT * FROM group_join_requests WHERE request_id=?",
                        (str(request_id),))
         row = cursor.fetchone()
diff --git a/gn3/auth/authorisation/groups/views.py b/gn3/auth/authorisation/groups/views.py
index cf99975..7b967d7 100644
--- a/gn3/auth/authorisation/groups/views.py
+++ b/gn3/auth/authorisation/groups/views.py
@@ -76,7 +76,7 @@ def request_to_join(group_id: uuid.UUID) -> Response:
     def __request__(conn: db.DbConnection, user: User, group_id: uuid.UUID,
                     message: str):
         with db.cursor(conn) as cursor:
-            group = user_group(cursor, user).maybe(# type: ignore[misc]
+            group = user_group(conn, user).maybe(# type: ignore[misc]
                 False, lambda grp: grp)# type: ignore[arg-type]
             if group:
                 error = AuthorisationError(
@@ -148,7 +148,7 @@ def unlinked_data(resource_type: str) -> Response:
     with require_oauth.acquire("profile group resource") as the_token:
         db_uri = current_app.config["AUTH_DB"]
         with db.connection(db_uri) as conn, db.cursor(conn) as cursor:
-            ugroup = user_group(cursor, the_token.user).maybe(# type: ignore[misc]
+            ugroup = user_group(conn, the_token.user).maybe(# type: ignore[misc]
                 DUMMY_GROUP, lambda grp: grp)
             if ugroup == DUMMY_GROUP:
                 return jsonify(tuple())
@@ -233,7 +233,7 @@ def group_roles():
         def __list_roles__(conn: db.DbConnection):
             ## TODO: Check that user has appropriate privileges
             with db.cursor(conn) as cursor:
-                group = user_group(cursor, the_token.user).maybe(# type: ignore[misc]
+                group = user_group(conn, the_token.user).maybe(# type: ignore[misc]
                     DUMMY_GROUP, lambda grp: grp)
                 if group == DUMMY_GROUP:
                     return tuple()
@@ -291,8 +291,7 @@ def create_group_role():
                 raise InvalidData(
                     "At least one privilege needs to be provided.")
 
-            with db.cursor(conn) as cursor:
-                group = user_group(cursor, the_token.user).maybe(# type: ignore[misc]
+            group = user_group(conn, the_token.user).maybe(# type: ignore[misc]
                     DUMMY_GROUP, lambda grp: grp)
 
             if group == DUMMY_GROUP:
@@ -314,9 +313,8 @@ def view_group_role(group_role_id: uuid.UUID):
     """Return the details of the given role."""
     with require_oauth.acquire("profile group role") as the_token:
         def __group_role__(conn: db.DbConnection) -> GroupRole:
-            with db.cursor(conn) as cursor:
-                group = user_group(cursor, the_token.user).maybe(#type: ignore[misc]
-                    DUMMY_GROUP, lambda grp: grp)
+            group = user_group(conn, the_token.user).maybe(#type: ignore[misc]
+                DUMMY_GROUP, lambda grp: grp)
 
             if group == DUMMY_GROUP:
                 raise AuthorisationError(
@@ -329,29 +327,28 @@ def __add_remove_priv_to_from_role__(conn: db.DbConnection,
                                      direction: str,
                                      user: User) -> GroupRole:
     assert direction in ("ADD", "DELETE")
-    with db.cursor(conn) as cursor:
-        group = user_group(cursor, user).maybe(# type: ignore[misc]
-            DUMMY_GROUP, lambda grp: grp)
-
-        if group == DUMMY_GROUP:
-            raise AuthorisationError(
-                "You need to be a member of a group to edit roles.")
-        try:
-            privilege_id = request.form.get("privilege_id", "")
-            assert bool(privilege_id), "Privilege to add must be provided."
-            privileges = privileges_by_ids(conn, (privilege_id,))
-            if len(privileges) == 0:
-                raise NotFoundError("Privilege not found.")
-            dir_fns = {
-                "ADD": add_privilege_to_group_role,
-                "DELETE": delete_privilege_to_group_role
-            }
-            return dir_fns[direction](
-                conn,
-                group_role_by_id(conn, group, group_role_id),
-                privileges[0])
-        except AssertionError as aerr:
-            raise InvalidData(aerr.args[0]) from aerr
+    group = user_group(conn, user).maybe(# type: ignore[misc]
+        DUMMY_GROUP, lambda grp: grp)
+
+    if group == DUMMY_GROUP:
+        raise AuthorisationError(
+            "You need to be a member of a group to edit roles.")
+    try:
+        privilege_id = request.form.get("privilege_id", "")
+        assert bool(privilege_id), "Privilege to add must be provided."
+        privileges = privileges_by_ids(conn, (privilege_id,))
+        if len(privileges) == 0:
+            raise NotFoundError("Privilege not found.")
+        dir_fns = {
+            "ADD": add_privilege_to_group_role,
+            "DELETE": delete_privilege_to_group_role
+        }
+        return dir_fns[direction](
+            conn,
+            group_role_by_id(conn, group, group_role_id),
+            privileges[0])
+    except AssertionError as aerr:
+        raise InvalidData(aerr.args[0]) from aerr
 
 @groups.route("/role/<uuid:group_role_id>/privilege/add", methods=["POST"])
 @require_oauth("profile group")
diff --git a/gn3/auth/authorisation/resources/models.py b/gn3/auth/authorisation/resources/models.py
index d0dd2f4..4049fae 100644
--- a/gn3/auth/authorisation/resources/models.py
+++ b/gn3/auth/authorisation/resources/models.py
@@ -87,7 +87,7 @@ def create_resource(
         resource_category: ResourceCategory, user: User) -> Resource:
     """Create a resource item."""
     with db.cursor(conn) as cursor:
-        group = user_group(cursor, user).maybe(
+        group = user_group(conn, user).maybe(
             False, lambda grp: grp)# type: ignore[misc, arg-type]
         if not group:
             raise MissingGroupError(
@@ -153,16 +153,17 @@ def public_resources(conn: db.DbConnection) -> Sequence[Resource]:
             for row in results)
 
 def group_leader_resources(
-        cursor: db.DbCursor, user: User, group: Group,
+        conn: db.DbConnection, user: User, group: Group,
         res_categories: Dict[UUID, ResourceCategory]) -> Sequence[Resource]:
     """Return all the resources available to the group leader"""
-    if is_group_leader(cursor, user, group):
-        cursor.execute("SELECT * FROM resources WHERE group_id=?",
-                       (str(group.group_id),))
-        return tuple(
-            Resource(group, UUID(row[1]), row[2], res_categories[UUID(row[3])],
-                     bool(row[4]))
-            for row in cursor.fetchall())
+    if is_group_leader(conn, user, group):
+        with db.cursor(conn) as cursor:
+            cursor.execute("SELECT * FROM resources WHERE group_id=?",
+                           (str(group.group_id),))
+            return tuple(
+                Resource(group, UUID(row[1]), row[2],
+                         res_categories[UUID(row[3])], bool(row[4]))
+                for row in cursor.fetchall())
     return tuple()
 
 def user_resources(conn: db.DbConnection, user: User) -> Sequence[Resource]:
@@ -172,7 +173,7 @@ def user_resources(conn: db.DbConnection, user: User) -> Sequence[Resource]:
     }
     with db.cursor(conn) as cursor:
         def __all_resources__(group) -> Sequence[Resource]:
-            gl_resources = group_leader_resources(cursor, user, group, categories)
+            gl_resources = group_leader_resources(conn, user, group, categories)
 
             cursor.execute(
                 ("SELECT resources.* FROM group_user_roles_on_resources "
@@ -193,7 +194,7 @@ def user_resources(conn: db.DbConnection, user: User) -> Sequence[Resource]:
             }.values())
 
         # Fix the typing here
-        return user_group(cursor, user).map(__all_resources__).maybe(# type: ignore[arg-type,misc]
+        return user_group(conn, user).map(__all_resources__).maybe(# type: ignore[arg-type,misc]
             public_resources(conn), lambda res: res)# type: ignore[arg-type,return-value]
 
 def attach_resource_data(cursor: db.DbCursor, resource: Resource) -> Resource:
diff --git a/gn3/auth/authorisation/users/views.py b/gn3/auth/authorisation/users/views.py
index e3901b4..f343e77 100644
--- a/gn3/auth/authorisation/users/views.py
+++ b/gn3/auth/authorisation/users/views.py
@@ -34,8 +34,8 @@ def user_details() -> Response:
             "user_id": user.user_id, "email": user.email, "name": user.name,
             "group": False
         }
-        with db.connection(current_app.config["AUTH_DB"]) as conn, db.cursor(conn) as cursor:
-            the_group = _user_group(cursor, user).maybe(# type: ignore[misc]
+        with db.connection(current_app.config["AUTH_DB"]) as conn:
+            the_group = _user_group(conn, user).maybe(# type: ignore[misc]
                 False, lambda grp: grp)# type: ignore[arg-type]
             return jsonify({
                 **user_dets,
@@ -61,7 +61,8 @@ def validate_password(password, confirm_password) -> str:
 
     return password
 
-def __valid_username__(name: str) -> str:
+def validate_username(name: str) -> str:
+    """Validate the provides name."""
     if name == "":
         raise UsernameError("User's name not provided.")
 
@@ -89,7 +90,7 @@ def register_user() -> Response:
             password = validate_password(
                 form.get("password", "").strip(),
                 form.get("confirm_password", "").strip())
-            user_name = __valid_username__(form.get("user_name", "").strip())
+            user_name = validate_username(form.get("user_name", "").strip())
             with db.cursor(conn) as cursor:
                 user, _hashed_password = set_user_password(
                     cursor, save_user(
@@ -118,8 +119,8 @@ def user_group() -> Response:
     """Retrieve the group in which the user is a member."""
     with require_oauth.acquire("profile group") as the_token:
         db_uri = current_app.config["AUTH_DB"]
-        with db.connection(db_uri) as conn, db.cursor(conn) as cursor:
-            group = _user_group(cursor, the_token.user).maybe(# type: ignore[misc]
+        with db.connection(db_uri) as conn:
+            group = _user_group(conn, the_token.user).maybe(# type: ignore[misc]
                 False, lambda grp: grp)# type: ignore[arg-type]
 
         if group: