""" Assign any existing data (that is not currently assigned to any group) to the default sys-admin group for accessibility purposes. """ import sys import json import time import random import logging from pathlib import Path from uuid import UUID, uuid4 import click from gn_libs import mysqldb as biodb from MySQLdb.cursors import DictCursor import gn_auth.auth.db.sqlite3 as authdb from gn_auth.auth.authentication.users import User from gn_auth.auth.authorisation.roles.models import ( revoke_user_role_by_name, assign_user_role_by_name) from gn_auth.auth.authorisation.resources.groups.models import ( Group, save_group, add_resources_to_group) from gn_auth.auth.authorisation.resources.common import assign_resource_owner_role from gn_auth.auth.authorisation.resources.models import Resource, ResourceCategory class DataNotFound(Exception): """Raise if no admin user exists.""" def sys_admins(conn: authdb.DbConnection) -> tuple[User, ...]: """Retrieve all the existing system admins.""" with authdb.cursor(conn) as cursor: cursor.execute( "SELECT u.* FROM users AS u " "INNER JOIN user_roles AS ur ON u.user_id=ur.user_id " "INNER JOIN roles AS r ON ur.role_id=r.role_id " "WHERE r.role_name='system-administrator'") return tuple(User.from_sqlite3_row(row) for row in cursor.fetchall()) return tuple() def choose_admin(enum_admins: dict[int, User]) -> int: """Prompt and read user choice.""" while True: try: print("\n===========================\n") print("We found the following system administrators:") for idx, admin in enum_admins.items(): print(f"\t{idx}: {admin.name} ({admin.email})") choice = input(f"Choose [1 .. {len(enum_admins)}]: ") return int(choice) except ValueError as _verr: if choice.lower() == "quit": print("Goodbye!") sys.exit(0) print(f"\nERROR: Invalid choice '{choice}'!") def select_sys_admin(admins: tuple[User, ...]) -> User: """Pick one admin out of list.""" if len(admins) > 0: if len(admins) == 1: print(f"-> Found Admin: {admins[0].name} ({admins[0].email})") return admins[0] enum_admins = dict(enumerate(admins, start=1)) chosen = enum_admins[choose_admin(enum_admins)] print(f"-> Chosen Admin: {chosen.name} ({chosen.email})") return chosen raise DataNotFound( "No administrator user found. Create an administrator user first.") def admin_group(conn: authdb.DbConnection, admin: User) -> Group: """Retrieve the admin's user group. If none exist, create one.""" with authdb.cursor(conn) as cursor: cursor.execute( "SELECT g.* FROM users AS u " "INNER JOIN group_users AS gu ON u.user_id=gu.user_id " "INNER JOIN groups AS g on gu.group_id=g.group_id " "WHERE u.user_id = ?", (str(admin.user_id),)) row = cursor.fetchone() if row: return Group(UUID(row["group_id"]), row["group_name"], json.loads(row["group_metadata"])) new_group = save_group(cursor, "AutoAdminGroup", { "group_description": ( "Created by script for existing data visibility. " "Existing data was migrated into this group and assigned " "to publicly visible resources according to type.") }) cursor.execute( "SELECT * FROM resource_categories WHERE " "resource_category_key='group'") res_cat_id = cursor.fetchone()["resource_category_id"] grp_res = { "group_id": str(new_group.group_id), "resource_id": str(uuid4()), "resource_name": new_group.group_name, "resource_category_id": res_cat_id, "public": 0 } cursor.execute( "INSERT INTO resources VALUES " "(:resource_id, :resource_name, :resource_category_id, :public)", grp_res) cursor.execute( "INSERT INTO group_resources(resource_id, group_id) " "VALUES(:resource_id, :group_id)", grp_res) cursor.execute("INSERT INTO group_users VALUES (?, ?)", (str(new_group.group_id), str(admin.user_id))) revoke_user_role_by_name(cursor, admin, "group-creator") assign_user_role_by_name( cursor, admin, UUID(grp_res["resource_id"]), "group-leader") return new_group def __resource_category_by_key__( cursor: authdb.DbCursor, category_key: str) -> ResourceCategory: """Retrieve a resource category by its ID.""" cursor.execute( "SELECT * FROM resource_categories WHERE resource_category_key = ?", (category_key,)) row = cursor.fetchone() if not bool(row): raise DataNotFound( f"Could not find resource category with key {category_key}") return ResourceCategory(UUID(row["resource_category_id"]), row["resource_category_key"], row["resource_category_description"]) def __create_resources__(cursor: authdb.DbCursor) -> tuple[Resource, ...]: """Create default resources.""" resources = tuple(Resource( uuid4(), name, __resource_category_by_key__(cursor, catkey), True, tuple() ) for name, catkey in ( ("mRNA-euhrin", "mrna"), ("pheno-xboecp", "phenotype"), ("geno-welphd", "genotype"))) cursor.executemany( "INSERT INTO resources VALUES (:rid, :rname, :rcid, :pub)", tuple({ "rid": str(res.resource_id), "rname": res.resource_name, "rcid": str(res.resource_category.resource_category_id), "pub": 1 } for res in resources)) return resources def default_resources(conn: authdb.DbConnection, group: Group) -> tuple[ Resource, ...]: """Create default resources, or return them if they exist.""" with authdb.cursor(conn) as cursor: cursor.execute( "SELECT r.resource_id, r.resource_name, r.public, rc.* " "FROM resource_ownership AS ro INNER JOIN resources AS r " "ON ro.resource_id=r.resource_id " "INNER JOIN resource_categories AS rc " "ON r.resource_category_id=rc.resource_category_id " "WHERE ro.group_id=? AND r.resource_name IN " "('mRNA-euhrin', 'pheno-xboecp', 'geno-welphd')", (str(group.group_id),)) rows = cursor.fetchall() if len(rows) == 0: return __create_resources__(cursor) return tuple(Resource( UUID(row["resource_id"]), row["resource_name"], ResourceCategory( UUID(row["resource_category_id"]), row["resource_category_key"], row["resource_category_description"]), bool(row["public"]), tuple() ) for row in rows) def delay(): """Delay a while: anything from 2 seconds to 15 seconds.""" time.sleep(random.choice(range(2,16))) def __assigned_mrna__(authconn): """Retrieve assigned mRNA items.""" with authdb.cursor(authconn) as cursor: cursor.execute( "SELECT SpeciesId, InbredSetId, ProbeFreezeId, ProbeSetFreezeId " "FROM linked_mrna_data") return tuple( (row["SpeciesId"], row["InbredSetId"], row["ProbeFreezeId"], row["ProbeSetFreezeId"]) for row in cursor.fetchall()) def __unassigned_mrna__(bioconn, assigned): """Retrieve unassigned mRNA data items.""" query = ( "SELECT s.SpeciesId, iset.InbredSetId, pf.ProbeFreezeId, " "psf.Id AS ProbeSetFreezeId, psf.Name AS dataset_name, " "psf.FullName AS dataset_fullname, psf.ShortName AS dataset_shortname " "FROM Species AS s INNER JOIN InbredSet AS iset " "ON s.SpeciesId=iset.SpeciesId INNER JOIN ProbeFreeze AS pf " "ON iset.InbredSetId=pf.InbredSetId INNER JOIN ProbeSetFreeze AS psf " "ON pf.ProbeFreezeId=psf.ProbeFreezeId " "WHERE s.Name != 'human' ") if len(assigned) > 0: paramstr = ", ".join(["(%s, %s, %s, %s)"] * len(assigned)) query = query + ( "AND (s.SpeciesId, iset.InbredSetId, pf.ProbeFreezeId, psf.Id) " f"NOT IN ({paramstr}) ") query = query + "LIMIT 100000" with bioconn.cursor(DictCursor) as cursor: cursor.execute(query, tuple(item for row in assigned for item in row)) return (row for row in cursor.fetchall()) def __assign_mrna__(authconn, bioconn, resource, group): "Assign any unassigned mRNA data to resource." while True: unassigned = tuple({ "data_link_id": str(uuid4()), "group_id": str(group.group_id), "resource_id": str(resource.resource_id), **row } for row in __unassigned_mrna__( bioconn, __assigned_mrna__(authconn))) if len(unassigned) <= 0: print("-> mRNA: Completed!") break with authdb.cursor(authconn) as cursor: cursor.executemany( "INSERT INTO linked_mrna_data VALUES " "(:data_link_id, :group_id, :SpeciesId, :InbredSetId, " ":ProbeFreezeId, :ProbeSetFreezeId, :dataset_name, " ":dataset_fullname, :dataset_shortname)", unassigned) cursor.executemany( "INSERT INTO mrna_resources VALUES " "(:resource_id, :data_link_id)", unassigned) print(f"-> mRNA: Linked {len(unassigned)}") delay() def __assigned_geno__(authconn): """Retrieve assigned genotype data.""" with authdb.cursor(authconn) as cursor: cursor.execute( "SELECT SpeciesId, InbredSetId, GenoFreezeId " "FROM linked_genotype_data") return tuple((row["SpeciesId"], row["InbredSetId"], row["GenoFreezeId"]) for row in cursor.fetchall()) def __unassigned_geno__(bioconn, assigned): """Fetch unassigned genotype data.""" query = ( "SELECT s.SpeciesId, iset.InbredSetId, iset.InbredSetName, " "gf.Id AS GenoFreezeId, gf.Name AS dataset_name, " "gf.FullName AS dataset_fullname, " "gf.ShortName AS dataset_shortname " "FROM Species AS s INNER JOIN InbredSet AS iset " "ON s.SpeciesId=iset.SpeciesId INNER JOIN GenoFreeze AS gf " "ON iset.InbredSetId=gf.InbredSetId " "WHERE s.Name != 'human' ") if len(assigned) > 0: paramstr = ", ".join(["(%s, %s, %s)"] * len(assigned)) query = query + ( "AND (s.SpeciesId, iset.InbredSetId, gf.Id) " f"NOT IN ({paramstr}) ") query = query + "LIMIT 100000" with bioconn.cursor(DictCursor) as cursor: cursor.execute(query, tuple(item for row in assigned for item in row)) return (row for row in cursor.fetchall()) def __assign_geno__(authconn, bioconn, resource, group): "Assign any unassigned Genotype data to resource." while True: unassigned = tuple({ "data_link_id": str(uuid4()), "group_id": str(group.group_id), "resource_id": str(resource.resource_id), **row } for row in __unassigned_geno__( bioconn, __assigned_geno__(authconn))) if len(unassigned) <= 0: print("-> Genotype: Completed!") break with authdb.cursor(authconn) as cursor: cursor.executemany( "INSERT INTO linked_genotype_data VALUES " "(:data_link_id, :group_id, :SpeciesId, :InbredSetId, " ":GenoFreezeId, :dataset_name, :dataset_fullname, " ":dataset_shortname)", unassigned) cursor.executemany( "INSERT INTO genotype_resources VALUES " "(:resource_id, :data_link_id)", unassigned) print(f"-> Genotype: Linked {len(unassigned)}") delay() def __assigned_pheno__(authconn): """Retrieve assigned phenotype data.""" with authdb.cursor(authconn) as cursor: cursor.execute( "SELECT SpeciesId, InbredSetId, PublishFreezeId, PublishXRefId " "FROM linked_phenotype_data") return tuple(( row["SpeciesId"], row["InbredSetId"], row["PublishFreezeId"], row["PublishXRefId"]) for row in cursor.fetchall()) def __unassigned_pheno__(bioconn, assigned): """Retrieve all unassigned Phenotype data.""" query = ( "SELECT spc.SpeciesId, iset.InbredSetId, " "pf.Id AS PublishFreezeId, pf.Name AS dataset_name, " "pf.FullName AS dataset_fullname, " "pf.ShortName AS dataset_shortname, pxr.Id AS PublishXRefId " "FROM " "Species AS spc " "INNER JOIN InbredSet AS iset " "ON spc.SpeciesId=iset.SpeciesId " "INNER JOIN PublishFreeze AS pf " "ON iset.InbredSetId=pf.InbredSetId " "INNER JOIN PublishXRef AS pxr " "ON pf.InbredSetId=pxr.InbredSetId " "WHERE spc.Name != 'human' ") if len(assigned) > 0: paramstr = ", ".join(["(%s, %s, %s, %s)"] * len(assigned)) query = query + ( "AND (spc.SpeciesId, iset.InbredSetId, pf.Id, pxr.Id) " f"NOT IN ({paramstr}) ") query = query + "LIMIT 100000" with bioconn.cursor(DictCursor) as cursor: cursor.execute(query, tuple(item for row in assigned for item in row)) return (row for row in cursor.fetchall()) def __assign_pheno__(authconn, bioconn, resource, group): """Assign any unassigned Phenotype data to resource.""" while True: unassigned = tuple({ "data_link_id": str(uuid4()), "group_id": str(group.group_id), "resource_id": str(resource.resource_id), **row } for row in __unassigned_pheno__( bioconn, __assigned_pheno__(authconn))) if len(unassigned) <= 0: print("-> Phenotype: Completed!") break with authdb.cursor(authconn) as cursor: cursor.executemany( "INSERT INTO linked_phenotype_data VALUES " "(:data_link_id, :group_id, :SpeciesId, :InbredSetId, " ":PublishFreezeId, :dataset_name, :dataset_fullname, " ":dataset_shortname, :PublishXRefId)", unassigned) cursor.executemany( "INSERT INTO phenotype_resources VALUES " "(:resource_id, :data_link_id)", unassigned) print(f"-> Phenotype: Linked {len(unassigned)}") delay() def assign_data_to_resource( authconn, bioconn, resource: Resource, group: Group): """Assign existing data, not linked to any group to the resource.""" assigner_fns = { "mrna": __assign_mrna__, "genotype": __assign_geno__, "phenotype": __assign_pheno__ } return assigner_fns[resource.resource_category.resource_category_key]( authconn, bioconn, resource, group) def entry(authdbpath, mysqldburi): """Entry-point for data migration.""" if not Path(authdbpath).exists(): print( f"ERROR: Auth db file `{authdbpath}` does not exist.", file=sys.stderr) sys.exit(2) try: with (authdb.connection(authdbpath) as authconn, biodb.database_connection(mysqldburi) as bioconn): admin = select_sys_admin(sys_admins(authconn)) the_admin_group = admin_group(authconn, admin) resources = default_resources(authconn, the_admin_group) add_resources_to_group(authconn, resources, the_admin_group) for resource in resources: assign_data_to_resource( authconn, bioconn, resource, the_admin_group) with authdb.cursor(authconn) as cursor: assign_resource_owner_role( cursor, resource.resource_id, admin.user_id) except DataNotFound as dnf: print(dnf.args[0], file=sys.stderr) sys.exit(1) @click.command() @click.argument("authdbpath") # "Path to the Auth(entic|oris)ation database" @click.argument("mysqldburi") # "URI to the MySQL database with the biology data" @click.option("--loglevel", default="WARNING", show_default=True, type=click.Choice(["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"])) def run(authdbpath, mysqldburi, loglevel): """Setup command-line arguments.""" globallogger = logging.getLogger() globallogger.setLevel(loglevel) entry(authdbpath, mysqldburi) if __name__ == "__main__": run() # pylint: disable=[no-value-for-parameter]