diff options
Diffstat (limited to 'gn3/auth/authorisation/data')
| -rw-r--r-- | gn3/auth/authorisation/data/__init__.py | 0 | ||||
| -rw-r--r-- | gn3/auth/authorisation/data/genotypes.py | 96 | ||||
| -rw-r--r-- | gn3/auth/authorisation/data/mrna.py | 100 | ||||
| -rw-r--r-- | gn3/auth/authorisation/data/phenotypes.py | 140 | ||||
| -rw-r--r-- | gn3/auth/authorisation/data/views.py | 310 |
5 files changed, 0 insertions, 646 deletions
diff --git a/gn3/auth/authorisation/data/__init__.py b/gn3/auth/authorisation/data/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/gn3/auth/authorisation/data/__init__.py +++ /dev/null diff --git a/gn3/auth/authorisation/data/genotypes.py b/gn3/auth/authorisation/data/genotypes.py deleted file mode 100644 index 8f901a5..0000000 --- a/gn3/auth/authorisation/data/genotypes.py +++ /dev/null @@ -1,96 +0,0 @@ -"""Handle linking of Genotype data to the Auth(entic|oris)ation system.""" -import uuid -from typing import Iterable - -from MySQLdb.cursors import DictCursor - -import gn3.auth.db as authdb -import gn3.db_utils as gn3db -from gn3.auth.dictify import dictify -from gn3.auth.authorisation.checks import authorised_p -from gn3.auth.authorisation.groups.models import Group - -def linked_genotype_data(conn: authdb.DbConnection) -> Iterable[dict]: - """Retrive genotype data that is linked to user groups.""" - with authdb.cursor(conn) as cursor: - cursor.execute("SELECT * FROM linked_genotype_data") - return (dict(row) for row in cursor.fetchall()) - -@authorised_p(("system:data:link-to-group",), - error_description=( - "You do not have sufficient privileges to link data to (a) " - "group(s)."), - oauth2_scope="profile group resource") -def ungrouped_genotype_data(# pylint: disable=[too-many-arguments] - authconn: authdb.DbConnection, gn3conn: gn3db.Connection, - search_query: str, selected: tuple[dict, ...] = tuple(), - limit: int = 10000, offset: int = 0) -> tuple[ - dict, ...]: - """Retrieve genotype data that is not linked to any user group.""" - params = tuple( - (row["SpeciesId"], row["InbredSetId"], row["GenoFreezeId"]) - for row in linked_genotype_data(authconn)) + tuple( - (row["SpeciesId"], row["InbredSetId"], row["GenoFreezeId"]) - for row in selected) - 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 ") - - if len(params) > 0 or bool(search_query): - query = query + "WHERE " - - if len(params) > 0: - paramstr = ", ".join(["(%s, %s, %s)"] * len(params)) - query = query + ( - "(s.SpeciesId, iset.InbredSetId, gf.Id) " - f"NOT IN ({paramstr}) " - ) + ("AND " if bool(search_query) else "") - - if bool(search_query): - query = query + ( - "CONCAT(gf.Name, ' ', gf.FullName, ' ', gf.ShortName) LIKE %s ") - params = params + ((f"%{search_query}%",),)# type: ignore[operator] - - query = query + f"LIMIT {int(limit)} OFFSET {int(offset)}" - with gn3conn.cursor(DictCursor) as cursor: - cursor.execute( - query, tuple(item for sublist in params for item in sublist)) - return tuple(row for row in cursor.fetchall()) - -@authorised_p( - ("system:data:link-to-group",), - error_description=( - "You do not have sufficient privileges to link data to (a) " - "group(s)."), - oauth2_scope="profile group resource") -def link_genotype_data( - conn: authdb.DbConnection, group: Group, datasets: dict) -> dict: - """Link genotye `datasets` to `group`.""" - with authdb.cursor(conn) as cursor: - cursor.executemany( - "INSERT INTO linked_genotype_data VALUES " - "(:data_link_id, :group_id, :SpeciesId, :InbredSetId, " - ":GenoFreezeId, :dataset_name, :dataset_fullname, " - ":dataset_shortname) " - "ON CONFLICT (SpeciesId, InbredSetId, GenoFreezeId) DO NOTHING", - tuple({ - "data_link_id": str(uuid.uuid4()), - "group_id": str(group.group_id), - **{ - key: value for key,value in dataset.items() if key in ( - "GenoFreezeId", "InbredSetId", "SpeciesId", - "dataset_fullname", "dataset_name", "dataset_shortname") - } - } for dataset in datasets)) - return { - "description": ( - f"Successfully linked {len(datasets)} to group " - f"'{group.group_name}'."), - "group": dictify(group), - "datasets": datasets - } diff --git a/gn3/auth/authorisation/data/mrna.py b/gn3/auth/authorisation/data/mrna.py deleted file mode 100644 index bdfc5c1..0000000 --- a/gn3/auth/authorisation/data/mrna.py +++ /dev/null @@ -1,100 +0,0 @@ -"""Handle linking of mRNA Assay data to the Auth(entic|oris)ation system.""" -import uuid -from typing import Iterable -from MySQLdb.cursors import DictCursor - -import gn3.auth.db as authdb -import gn3.db_utils as gn3db -from gn3.auth.dictify import dictify -from gn3.auth.authorisation.checks import authorised_p -from gn3.auth.authorisation.groups.models import Group - -def linked_mrna_data(conn: authdb.DbConnection) -> Iterable[dict]: - """Retrieve mRNA Assay data that is linked to user groups.""" - with authdb.cursor(conn) as cursor: - cursor.execute("SELECT * FROM linked_mrna_data") - return (dict(row) for row in cursor.fetchall()) - -@authorised_p(("system:data:link-to-group",), - error_description=( - "You do not have sufficient privileges to link data to (a) " - "group(s)."), - oauth2_scope="profile group resource") -def ungrouped_mrna_data(# pylint: disable=[too-many-arguments] - authconn: authdb.DbConnection, gn3conn: gn3db.Connection, - search_query: str, selected: tuple[dict, ...] = tuple(), - limit: int = 10000, offset: int = 0) -> tuple[ - dict, ...]: - """Retrieve mrna data that is not linked to any user group.""" - params = tuple( - (row["SpeciesId"], row["InbredSetId"], row["ProbeFreezeId"], - row["ProbeSetFreezeId"]) - for row in linked_mrna_data(authconn)) + tuple( - (row["SpeciesId"], row["InbredSetId"], row["ProbeFreezeId"], - row["ProbeSetFreezeId"]) - for row in selected) - query = ( - "SELECT s.SpeciesId, iset.InbredSetId, iset.InbredSetName, " - "pf.ProbeFreezeId, pf.Name AS StudyName, 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 " if (len(params) > 0 or bool(search_query)) else "") - - if len(params) > 0: - paramstr = ", ".join(["(%s, %s, %s, %s)"] * len(params)) - query = query + ( - "(s.SpeciesId, iset.InbredSetId, pf.ProbeFreezeId, psf.Id) " - f"NOT IN ({paramstr}) " - ) + ("AND " if bool(search_query) else "") - - if bool(search_query): - query = query + ( - "CONCAT(pf.Name, psf.Name, ' ', psf.FullName, ' ', psf.ShortName) " - "LIKE %s ") - params = params + ((f"%{search_query}%",),)# type: ignore[operator] - - query = query + f"LIMIT {int(limit)} OFFSET {int(offset)}" - with gn3conn.cursor(DictCursor) as cursor: - cursor.execute( - query, tuple(item for sublist in params for item in sublist)) - return tuple(row for row in cursor.fetchall()) - -@authorised_p( - ("system:data:link-to-group",), - error_description=( - "You do not have sufficient privileges to link data to (a) " - "group(s)."), - oauth2_scope="profile group resource") -def link_mrna_data( - conn: authdb.DbConnection, group: Group, datasets: dict) -> dict: - """Link genotye `datasets` to `group`.""" - with authdb.cursor(conn) 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) " - "ON CONFLICT " - "(SpeciesId, InbredSetId, ProbeFreezeId, ProbeSetFreezeId) " - "DO NOTHING", - tuple({ - "data_link_id": str(uuid.uuid4()), - "group_id": str(group.group_id), - **{ - key: value for key,value in dataset.items() if key in ( - "SpeciesId", "InbredSetId", "ProbeFreezeId", - "ProbeSetFreezeId", "dataset_fullname", "dataset_name", - "dataset_shortname") - } - } for dataset in datasets)) - return { - "description": ( - f"Successfully linked {len(datasets)} to group " - f"'{group.group_name}'."), - "group": dictify(group), - "datasets": datasets - } diff --git a/gn3/auth/authorisation/data/phenotypes.py b/gn3/auth/authorisation/data/phenotypes.py deleted file mode 100644 index ff98295..0000000 --- a/gn3/auth/authorisation/data/phenotypes.py +++ /dev/null @@ -1,140 +0,0 @@ -"""Handle linking of Phenotype data to the Auth(entic|oris)ation system.""" -import uuid -from typing import Any, Iterable - -from MySQLdb.cursors import DictCursor - -import gn3.auth.db as authdb -import gn3.db_utils as gn3db -from gn3.auth.dictify import dictify -from gn3.auth.authorisation.checks import authorised_p -from gn3.auth.authorisation.groups.models import Group - -def linked_phenotype_data( - authconn: authdb.DbConnection, gn3conn: gn3db.Connection, - species: str = "") -> Iterable[dict[str, Any]]: - """Retrieve phenotype data linked to user groups.""" - authkeys = ("SpeciesId", "InbredSetId", "PublishFreezeId", "PublishXRefId") - with (authdb.cursor(authconn) as authcursor, - gn3conn.cursor(DictCursor) as gn3cursor): - authcursor.execute("SELECT * FROM linked_phenotype_data") - linked = tuple(tuple(row[key] for key in authkeys) - for row in authcursor.fetchall()) - if len(linked) <= 0: - return iter(()) - paramstr = ", ".join(["(%s, %s, %s, %s)"] * len(linked)) - query = ( - "SELECT spc.SpeciesId, spc.Name AS SpeciesName, iset.InbredSetId, " - "iset.InbredSetName, 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" if (len(linked) > 0 or bool(species)) else "") + ( - (" (spc.SpeciesId, iset.InbredSetId, pf.Id, pxr.Id) " - f"IN ({paramstr})") if len(linked) > 0 else "") + ( - " AND"if len(linked) > 0 else "") + ( - " spc.SpeciesName=%s" if bool(species) else "") - params = tuple(item for sublist in linked for item in sublist) + ( - (species,) if bool(species) else tuple()) - gn3cursor.execute(query, params) - return (item for item in gn3cursor.fetchall()) - -@authorised_p(("system:data:link-to-group",), - error_description=( - "You do not have sufficient privileges to link data to (a) " - "group(s)."), - oauth2_scope="profile group resource") -def ungrouped_phenotype_data( - authconn: authdb.DbConnection, gn3conn: gn3db.Connection): - """Retrieve phenotype data that is not linked to any user group.""" - with gn3conn.cursor() as cursor: - params = tuple( - (row["SpeciesId"], row["InbredSetId"], row["PublishFreezeId"], - row["PublishXRefId"]) - for row in linked_phenotype_data(authconn, gn3conn)) - paramstr = ", ".join(["(?, ?, ?, ?)"] * len(params)) - query = ( - "SELECT spc.SpeciesId, spc.SpeciesName, iset.InbredSetId, " - "iset.InbredSetName, 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") - if len(params) > 0: - query = query + ( - f" WHERE (iset.InbredSetId, pf.Id, pxr.Id) NOT IN ({paramstr})") - - cursor.execute(query, params) - return tuple(dict(row) for row in cursor.fetchall()) - - return tuple() - -def __traits__(gn3conn: gn3db.Connection, params: tuple[dict, ...]) -> tuple[dict, ...]: - """An internal utility function. Don't use outside of this module.""" - if len(params) < 1: - return tuple() - paramstr = ", ".join(["(%s, %s, %s, %s)"] * len(params)) - with gn3conn.cursor(DictCursor) as cursor: - cursor.execute( - "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.SpeciesName, iset.InbredSetName, pf.Name, pxr.Id) " - f"IN ({paramstr})", - tuple( - itm for sublist in ( - (item["species"], item["group"], item["dataset"], item["name"]) - for item in params) - for itm in sublist)) - return cursor.fetchall() - -@authorised_p(("system:data:link-to-group",), - error_description=( - "You do not have sufficient privileges to link data to (a) " - "group(s)."), - oauth2_scope="profile group resource") -def link_phenotype_data( - authconn:authdb.DbConnection, gn3conn: gn3db.Connection, group: Group, - traits: tuple[dict, ...]) -> dict: - """Link phenotype traits to a user group.""" - with authdb.cursor(authconn) as cursor: - params = tuple({ - "data_link_id": str(uuid.uuid4()), - "group_id": str(group.group_id), - **item - } for item in __traits__(gn3conn, traits)) - cursor.executemany( - "INSERT INTO linked_phenotype_data " - "VALUES (" - ":data_link_id, :group_id, :SpeciesId, :InbredSetId, " - ":PublishFreezeId, :dataset_name, :dataset_fullname, " - ":dataset_shortname, :PublishXRefId" - ")", - params) - return { - "description": ( - f"Successfully linked {len(traits)} traits to group."), - "group": dictify(group), - "traits": params - } diff --git a/gn3/auth/authorisation/data/views.py b/gn3/auth/authorisation/data/views.py deleted file mode 100644 index 81811dd..0000000 --- a/gn3/auth/authorisation/data/views.py +++ /dev/null @@ -1,310 +0,0 @@ -"""Handle data endpoints.""" -import sys -import uuid -import json -from typing import Any -from functools import partial - -import redis -from MySQLdb.cursors import DictCursor -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 import jobs -from gn3.commands import run_async_cmd -from gn3.db.traits import build_trait_name - -from gn3.auth import db -from gn3.auth.db_utils import with_db_connection - -from gn3.auth.authorisation.checks import require_json -from gn3.auth.authorisation.errors import InvalidData, NotFoundError - -from gn3.auth.authorisation.groups.models import group_by_id - -from gn3.auth.authorisation.users.models import user_resource_roles - -from gn3.auth.authorisation.resources.checks import authorised_for -from gn3.auth.authorisation.resources.models import ( - user_resources, public_resources, attach_resources_data) - -from gn3.auth.authorisation.oauth2.resource_server import require_oauth - -from gn3.auth.authorisation.users import User -from gn3.auth.authorisation.data.phenotypes import link_phenotype_data -from gn3.auth.authorisation.data.mrna import link_mrna_data, ungrouped_mrna_data -from gn3.auth.authorisation.data.genotypes import ( - link_genotype_data, ungrouped_genotype_data) - -data = Blueprint("data", __name__) - -@data.route("species") -def list_species() -> Response: - """List all available species information.""" - with (gn3db.database_connection(app.config["SQL_URI"]) as gn3conn, - gn3conn.cursor(DictCursor) as cursor): - cursor.execute("SELECT * FROM Species") - return jsonify(tuple(dict(row) for row in cursor.fetchall())) - -@data.route("/authorisation", methods=["POST"]) -@require_json -def authorisation() -> Response: - """Retrive the authorisation level for datasets/traits for the user.""" - # Access endpoint with something like: - # curl -X POST http://127.0.0.1:8080/api/oauth2/data/authorisation \ - # -H "Content-Type: application/json" \ - # -d '{"traits": ["HC_M2_0606_P::1442370_at", "BXDGeno::01.001.695", - # "BXDPublish::10001"]}' - db_uri = app.config["AUTH_DB"] - privileges = {} - user = User(uuid.uuid4(), "anon@ymous.user", "Anonymous User") - with db.connection(db_uri) as auth_conn: - try: - with require_oauth.acquire("profile group resource") as the_token: - user = the_token.user - resources = attach_resources_data( - auth_conn, user_resources(auth_conn, the_token.user)) - resources_roles = user_resource_roles(auth_conn, the_token.user) - privileges = { - resource_id: tuple( - privilege.privilege_id - for roles in resources_roles[resource_id] - for privilege in roles.privileges)#("group:resource:view-resource",) - for resource_id, is_authorised - in authorised_for( - auth_conn, the_token.user, - ("group:resource:view-resource",), tuple( - resource.resource_id for resource in resources)).items() - if is_authorised - } - except _HTTPException as exc: - err_msg = json.loads(exc.body) - if err_msg["error"] == "missing_authorization": - resources = attach_resources_data( - auth_conn, public_resources(auth_conn)) - else: - raise exc from None - - def __gen_key__(resource, data_item): - if resource.resource_category.resource_category_key.lower() == "phenotype": - return ( - f"{resource.resource_category.resource_category_key.lower()}::" - f"{data_item['dataset_name']}::{data_item['PublishXRefId']}") - return ( - f"{resource.resource_category.resource_category_key.lower()}::" - f"{data_item['dataset_name']}") - - data_to_resource_map = { - __gen_key__(resource, data_item): resource.resource_id - for resource in resources - for data_item in resource.resource_data - } - privileges = { - **{ - resource.resource_id: ("system:resource:public-read",) - for resource in resources if resource.public - }, - **privileges} - - args = request.get_json() - traits_names = args["traits"] # type: ignore[index] - def __translate__(val): - return { - "Temp": "Temp", - "ProbeSet": "mRNA", - "Geno": "Genotype", - "Publish": "Phenotype" - }[val] - - def __trait_key__(trait): - dataset_type = __translate__(trait['db']['dataset_type']).lower() - dataset_name = trait["db"]["dataset_name"] - if dataset_type == "phenotype": - return f"{dataset_type}::{dataset_name}::{trait['trait_name']}" - return f"{dataset_type}::{dataset_name}" - - return jsonify(tuple( - { - "user": user._asdict(), - **{key:trait[key] for key in ("trait_fullname", "trait_name")}, - "dataset_name": trait["db"]["dataset_name"], - "dataset_type": __translate__(trait["db"]["dataset_type"]), - "resource_id": data_to_resource_map.get(__trait_key__(trait)), - "privileges": privileges.get( - data_to_resource_map.get( - __trait_key__(trait), - uuid.UUID("4afa415e-94cb-4189-b2c6-f9ce2b6a878d")), - tuple()) + ( - # Temporary traits do not exist in db: Set them - # as public-read - ("system:resource:public-read",) - if trait["db"]["dataset_type"] == "Temp" - else tuple()) - } for trait in - (build_trait_name(trait_fullname) - for trait_fullname in traits_names))) - -def __search_mrna__(): - query = __request_key__("query", "") - limit = int(__request_key__("limit", 10000)) - offset = int(__request_key__("offset", 0)) - with gn3db.database_connection(app.config["SQL_URI"]) as gn3conn: - __ungrouped__ = partial( - ungrouped_mrna_data, gn3conn=gn3conn, search_query=query, - selected=__request_key_list__("selected"), - limit=limit, offset=offset) - return jsonify(with_db_connection(__ungrouped__)) - -def __request_key__(key: str, default: Any = ""): - if bool(request.json): - return request.json.get(#type: ignore[union-attr] - key, request.args.get(key, request.form.get(key, default))) - return request.args.get(key, request.form.get(key, default)) - -def __request_key_list__(key: str, default: tuple[Any, ...] = tuple()): - if bool(request.json): - return (request.json.get(key,[])#type: ignore[union-attr] - or request.args.getlist(key) or request.form.getlist(key) - or list(default)) - return (request.args.getlist(key) - or request.form.getlist(key) or list(default)) - -def __search_genotypes__(): - query = __request_key__("query", "") - limit = int(__request_key__("limit", 10000)) - offset = int(__request_key__("offset", 0)) - with gn3db.database_connection(app.config["SQL_URI"]) as gn3conn: - __ungrouped__ = partial( - ungrouped_genotype_data, gn3conn=gn3conn, search_query=query, - selected=__request_key_list__("selected"), - limit=limit, offset=offset) - return jsonify(with_db_connection(__ungrouped__)) - -def __search_phenotypes__(): - # launch the external process to search for phenotypes - redisuri = app.config["REDIS_URI"] - with redis.Redis.from_url(redisuri, decode_responses=True) as redisconn: - job_id = uuid.uuid4() - selected = __request_key__("selected_traits", []) - command =[ - sys.executable, "-m", "scripts.search_phenotypes", - __request_key__("species_name"), - __request_key__("query"), - str(job_id), - f"--host={__request_key__('gn3_server_uri')}", - f"--auth-db-uri={app.config['AUTH_DB']}", - f"--gn3-db-uri={app.config['SQL_URI']}", - f"--redis-uri={redisuri}", - f"--per-page={__request_key__('per_page')}"] +( - [f"--selected={json.dumps(selected)}"] - if len(selected) > 0 else []) - jobs.create_job(redisconn, { - "job_id": job_id, "command": command, "status": "queued", - "search_results": tuple()}) - return jsonify({ - "job_id": job_id, - "command_id": run_async_cmd( - redisconn, app.config.get("REDIS_JOB_QUEUE"), command), - "command": command - }) - -@data.route("/search", methods=["GET"]) -@require_oauth("profile group resource") -def search_unlinked_data(): - """Search for various unlinked data.""" - dataset_type = request.json["dataset_type"] - search_fns = { - "mrna": __search_mrna__, - "genotype": __search_genotypes__, - "phenotype": __search_phenotypes__ - } - return search_fns[dataset_type]() - -@data.route("/search/phenotype/<uuid:job_id>", methods=["GET"]) -def pheno_search_results(job_id: uuid.UUID) -> Response: - """Get the search results from the external script""" - def __search_error__(err): - raise NotFoundError(err["error_description"]) - redisuri = app.config["REDIS_URI"] - with redis.Redis.from_url(redisuri, decode_responses=True) as redisconn: - return jobs.job(redisconn, job_id).either( - __search_error__, jsonify) - -@data.route("/link/genotype", methods=["POST"]) -def link_genotypes() -> Response: - """Link genotype data to group.""" - def __values__(form) -> dict[str, Any]: - if not bool(form.get("species_name", "").strip()): - raise InvalidData("Expected 'species_name' not provided.") - if not bool(form.get("group_id")): - raise InvalidData("Expected 'group_id' not provided.",) - try: - _group_id = uuid.UUID(form.get("group_id")) - except TypeError as terr: - raise InvalidData("Expected a UUID for 'group_id' value.") from terr - if not bool(form.get("selected")): - raise InvalidData("Expected at least one dataset to be provided.") - return { - "group_id": uuid.UUID(form.get("group_id")), - "datasets": form.get("selected") - } - - def __link__(conn: db.DbConnection, group_id: uuid.UUID, datasets: dict): - return link_genotype_data(conn, group_by_id(conn, group_id), datasets) - - return jsonify(with_db_connection( - partial(__link__, **__values__(request.json)))) - -@data.route("/link/mrna", methods=["POST"]) -def link_mrna() -> Response: - """Link mrna data to group.""" - def __values__(form) -> dict[str, Any]: - if not bool(form.get("species_name", "").strip()): - raise InvalidData("Expected 'species_name' not provided.") - if not bool(form.get("group_id")): - raise InvalidData("Expected 'group_id' not provided.",) - try: - _group_id = uuid.UUID(form.get("group_id")) - except TypeError as terr: - raise InvalidData("Expected a UUID for 'group_id' value.") from terr - if not bool(form.get("selected")): - raise InvalidData("Expected at least one dataset to be provided.") - return { - "group_id": uuid.UUID(form.get("group_id")), - "datasets": form.get("selected") - } - - def __link__(conn: db.DbConnection, group_id: uuid.UUID, datasets: dict): - return link_mrna_data(conn, group_by_id(conn, group_id), datasets) - - return jsonify(with_db_connection( - partial(__link__, **__values__(request.json)))) - -@data.route("/link/phenotype", methods=["POST"]) -def link_phenotype() -> Response: - """Link phenotype data to group.""" - def __values__(form): - if not bool(form.get("species_name", "").strip()): - raise InvalidData("Expected 'species_name' not provided.") - if not bool(form.get("group_id")): - raise InvalidData("Expected 'group_id' not provided.",) - try: - _group_id = uuid.UUID(form.get("group_id")) - except TypeError as terr: - raise InvalidData("Expected a UUID for 'group_id' value.") from terr - if not bool(form.get("selected")): - raise InvalidData("Expected at least one dataset to be provided.") - return { - "group_id": uuid.UUID(form["group_id"]), - "traits": form["selected"] - } - - with gn3db.database_connection(app.config["SQL_URI"]) as gn3conn: - def __link__(conn: db.DbConnection, group_id: uuid.UUID, - traits: tuple[dict, ...]) -> dict: - return link_phenotype_data( - conn, gn3conn, group_by_id(conn, group_id), traits) - - return jsonify(with_db_connection( - partial(__link__, **__values__(request.json)))) |