diff options
Diffstat (limited to 'gn3/auth/authorisation/resources/data.py')
-rw-r--r-- | gn3/auth/authorisation/resources/data.py | 182 |
1 files changed, 0 insertions, 182 deletions
diff --git a/gn3/auth/authorisation/resources/data.py b/gn3/auth/authorisation/resources/data.py deleted file mode 100644 index 93b8e1d..0000000 --- a/gn3/auth/authorisation/resources/data.py +++ /dev/null @@ -1,182 +0,0 @@ -"""Handles the resource objects' data.""" -from typing import Any, Sequence - -from MySQLdb.cursors import DictCursor - -from gn3 import db_utils as gn3db -from gn3.auth import db as authdb -from gn3.auth.authorisation.groups import Group -from gn3.auth.authorisation.checks import authorised_p -from gn3.auth.authorisation.errors import InvalidData, NotFoundError - -def __fetch_grouped_data__( - conn: authdb.DbConnection, dataset_type: str) -> Sequence[dict[str, Any]]: - """Retrieve ids for all data that are linked to groups in the auth db.""" - with authdb.cursor(conn) as cursor: - cursor.execute( - "SELECT dataset_type, dataset_or_trait_id FROM linked_group_data " - "WHERE LOWER(dataset_type)=?", - (dataset_type,)) - return tuple(dict(row) for row in cursor.fetchall()) - -def __fetch_ungrouped_mrna_data__( - conn: gn3db.Connection, grouped_data, - offset: int = 0) -> Sequence[dict]: - """Fetch ungrouped mRNA Assay data.""" - query = ("SELECT psf.Id, psf.Name, psf.FullName, " - "ifiles.GN_AccesionId AS accession_id FROM ProbeSetFreeze AS psf " - "INNER JOIN InfoFiles AS ifiles ON psf.Name=ifiles.InfoPageName") - params: tuple[Any, ...] = tuple() - if bool(grouped_data): - clause = ", ".join(["%s"] * len(grouped_data)) - query = f"{query} WHERE psf.Id NOT IN ({clause})" - params = tuple(item["dataset_or_trait_id"] for item in grouped_data) - - query = f"{query} LIMIT 100 OFFSET %s" - with conn.cursor(DictCursor) as cursor: - cursor.execute(query, (params + (offset,))) - return tuple(dict(row) for row in cursor.fetchall()) - -def __fetch_ungrouped_geno_data__( - conn: gn3db.Connection, grouped_data, - offset: int = 0) -> Sequence[dict]: - """Fetch ungrouped Genotype data.""" - query = ("SELECT gf.Id, gf.Name, gf.FullName, " - "ifiles.GN_AccesionId AS accession_id FROM GenoFreeze AS gf " - "INNER JOIN InfoFiles AS ifiles ON gf.Name=ifiles.InfoPageName") - params: tuple[Any, ...] = tuple() - if bool(grouped_data): - clause = ", ".join(["%s"] * len(grouped_data)) - query = f"{query} WHERE gf.Id NOT IN ({clause})" - params = tuple(item["dataset_or_trait_id"] for item in grouped_data) - - query = f"{query} LIMIT 100 OFFSET %s" - with conn.cursor(DictCursor) as cursor: - cursor.execute(query, (params + (offset,))) - return tuple(dict(row) for row in cursor.fetchall()) - -def __fetch_ungrouped_pheno_data__( - conn: gn3db.Connection, grouped_data, - offset: int = 0) -> Sequence[dict]: - """Fetch ungrouped Phenotype data.""" - query = ("SELECT pf.Id, pf.Name, pf.FullName, " - "ifiles.GN_AccesionId AS accession_id FROM PublishFreeze AS pf " - "INNER JOIN InfoFiles AS ifiles ON pf.Name=ifiles.InfoPageName") - params: tuple[Any, ...] = tuple() - if bool(grouped_data): - clause = ", ".join(["%s"] * len(grouped_data)) - query = f"{query} WHERE pf.Id NOT IN ({clause})" - params = tuple(item["dataset_or_trait_id"] for item in grouped_data) - - query = f"{query} LIMIT 100 OFFSET %s" - with conn.cursor(DictCursor) as cursor: - cursor.execute(query, (params + (offset,))) - return tuple(dict(row) for row in cursor.fetchall()) - -def __fetch_ungrouped_data__( - conn: gn3db.Connection, dataset_type: str, - ungrouped: Sequence[dict[str, Any]]) -> Sequence[dict[str, Any]]: - """Fetch any ungrouped data.""" - fetch_fns = { - "mrna": __fetch_ungrouped_mrna_data__, - "genotype": __fetch_ungrouped_geno_data__, - "phenotype": __fetch_ungrouped_pheno_data__ - } - return fetch_fns[dataset_type](conn, ungrouped) - -@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 retrieve_ungrouped_data( - authconn: authdb.DbConnection, - gn3conn: gn3db.Connection, - dataset_type: str) -> Sequence[dict]: - """Retrieve any data not linked to any group.""" - if dataset_type not in ("mrna", "genotype", "phenotype"): - raise InvalidData( - "Requested dataset type is invalid. Expected one of " - "'mrna', 'genotype' or 'phenotype'.") - grouped_data = __fetch_grouped_data__(authconn, dataset_type) - return __fetch_ungrouped_data__(gn3conn, dataset_type, grouped_data) - -def __fetch_mrna_data_by_id__(conn: gn3db.Connection, dataset_id: str) -> dict: - """Fetch mRNA Assay data by ID.""" - with conn.cursor(DictCursor) as cursor: - cursor.execute( - "SELECT psf.Id, psf.Name, psf.FullName, " - "ifiles.GN_AccesionId AS accession_id FROM ProbeSetFreeze AS psf " - "INNER JOIN InfoFiles AS ifiles ON psf.Name=ifiles.InfoPageName " - "WHERE psf.Id=%s", - (dataset_id,)) - res = cursor.fetchone() - if res: - return dict(res) - raise NotFoundError("Could not find mRNA Assay data with the given ID.") - -def __fetch_geno_data_by_id__(conn: gn3db.Connection, dataset_id: str) -> dict: - """Fetch genotype data by ID.""" - with conn.cursor(DictCursor) as cursor: - cursor.execute( - "SELECT gf.Id, gf.Name, gf.FullName, " - "ifiles.GN_AccesionId AS accession_id FROM GenoFreeze AS gf " - "INNER JOIN InfoFiles AS ifiles ON gf.Name=ifiles.InfoPageName " - "WHERE gf.Id=%s", - (dataset_id,)) - res = cursor.fetchone() - if res: - return dict(res) - raise NotFoundError("Could not find Genotype data with the given ID.") - -def __fetch_pheno_data_by_id__(conn: gn3db.Connection, dataset_id: str) -> dict: - """Fetch phenotype data by ID.""" - with conn.cursor(DictCursor) as cursor: - cursor.execute( - "SELECT pf.Id, pf.Name, pf.FullName, " - "ifiles.GN_AccesionId AS accession_id FROM PublishFreeze AS pf " - "INNER JOIN InfoFiles AS ifiles ON pf.Name=ifiles.InfoPageName " - "WHERE pf.Id=%s", - (dataset_id,)) - res = cursor.fetchone() - if res: - return dict(res) - raise NotFoundError( - "Could not find Phenotype/Publish data with the given ID.") - -def __fetch_data_by_id( - conn: gn3db.Connection, dataset_type: str, dataset_id: str) -> dict: - """Fetch data from MySQL by ID.""" - fetch_fns = { - "mrna": __fetch_mrna_data_by_id__, - "genotype": __fetch_geno_data_by_id__, - "phenotype": __fetch_pheno_data_by_id__ - } - return fetch_fns[dataset_type](conn, dataset_id) - -@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_data_to_group( - authconn: authdb.DbConnection, gn3conn: gn3db.Connection, - dataset_type: str, dataset_id: str, group: Group) -> dict: - """Link the given data to the specified group.""" - the_data = __fetch_data_by_id(gn3conn, dataset_type, dataset_id) - with authdb.cursor(authconn) as cursor: - params = { - "group_id": str(group.group_id), "dataset_type": { - "mrna": "mRNA", "genotype": "Genotype", - "phenotype": "Phenotype" - }[dataset_type], - "dataset_or_trait_id": dataset_id, "dataset_name": the_data["Name"], - "dataset_fullname": the_data["FullName"], - "accession_id": the_data["accession_id"] - } - cursor.execute( - "INSERT INTO linked_group_data VALUES" - "(:group_id, :dataset_type, :dataset_or_trait_id, :dataset_name, " - ":dataset_fullname, :accession_id)", - params) - return params |