"""Functions to handle the low-level details regarding populations auth.""" from uuid import UUID, uuid4 from typing import Sequence, Optional import sqlite3 import gn_auth.auth.db.sqlite3 as db from gn_auth.auth.authentication.users import User from gn_auth.auth.authorisation.resources.base import Resource def assign_inbredset_group_owner_role( cursor: sqlite3.Cursor, resource: Resource, user: User ) -> Resource: """ Assign `user` as `InbredSet Group Owner` is resource category is 'inbredset-group'. """ if resource.resource_category.resource_category_key == "inbredset-group": cursor.execute( "SELECT * FROM roles WHERE role_name='inbredset-group-owner'") role = cursor.fetchone() cursor.execute( "INSERT INTO user_roles " "VALUES(:user_id, :role_id, :resource_id) " "ON CONFLICT (user_id, role_id, resource_id) DO NOTHING", { "user_id": str(user.user_id), "role_id": str(role["role_id"]), "resource_id": str(resource.resource_id) }) return resource def link_data_to_resource(# pylint: disable=[too-many-arguments, too-many-positional-arguments] cursor: sqlite3.Cursor, resource_id: UUID, species_id: int, population_id: int, population_name: str, population_fullname: str ) -> dict: """Link a species population to a resource for auth purposes.""" params = { "resource_id": str(resource_id), "data_link_id": str(uuid4()), "species_id": species_id, "population_id": population_id, "population_name": population_name, "population_fullname": population_fullname } cursor.execute( "INSERT INTO linked_inbredset_groups " "VALUES(" " :data_link_id," " :species_id," " :population_id," " :population_name," " :population_fullname" ")", params) cursor.execute( "INSERT INTO inbredset_group_resources " "VALUES (:resource_id, :data_link_id)", params) return params def resource_data( cursor: db.DbCursor, resource_id: UUID, offset: int = 0, limit: Optional[int] = None) -> Sequence[sqlite3.Row]: """Fetch data linked to a inbred-set resource""" cursor.execute( ("SELECT * FROM inbredset_group_resources AS igr " "INNER JOIN linked_inbredset_groups AS lig " "ON igr.data_link_id=lig.data_link_id " "WHERE igr.resource_id=?") + ( f" LIMIT {limit} OFFSET {offset}" if bool(limit) else ""), (str(resource_id),)) return cursor.fetchall()