diff options
Diffstat (limited to 'gn_auth/auth/authorisation/resources/inbredset')
| -rw-r--r-- | gn_auth/auth/authorisation/resources/inbredset/models.py | 85 | ||||
| -rw-r--r-- | gn_auth/auth/authorisation/resources/inbredset/views.py | 127 |
2 files changed, 207 insertions, 5 deletions
diff --git a/gn_auth/auth/authorisation/resources/inbredset/models.py b/gn_auth/auth/authorisation/resources/inbredset/models.py new file mode 100644 index 0000000..2626f3e --- /dev/null +++ b/gn_auth/auth/authorisation/resources/inbredset/models.py @@ -0,0 +1,85 @@ +"""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() diff --git a/gn_auth/auth/authorisation/resources/inbredset/views.py b/gn_auth/auth/authorisation/resources/inbredset/views.py index 444c442..9603b5b 100644 --- a/gn_auth/auth/authorisation/resources/inbredset/views.py +++ b/gn_auth/auth/authorisation/resources/inbredset/views.py @@ -1,12 +1,56 @@ """Views for InbredSet resources.""" -from flask import jsonify, Response, Blueprint +import uuid + +from pymonad.either import Left, Right, Either +from flask import jsonify, Response, Blueprint, current_app as app + from gn_auth.auth.db import sqlite3 as db -from gn_auth.auth.db.sqlite3 import with_db_connection +from gn_auth.auth.errors import NotFoundError +from gn_auth.auth.requests import request_json +from gn_auth.auth.authentication.users import User +from gn_auth.auth.authentication.oauth2.resource_server import require_oauth +from gn_auth.auth.authorisation.resources.base import Resource, ResourceCategory +from gn_auth.auth.authorisation.resources.groups.models import (Group, + user_group, + admin_group) +from gn_auth.auth.authorisation.resources.models import ( + create_resource as _create_resource) + +from .models import (link_data_to_resource, + assign_inbredset_group_owner_role) + +popbp = Blueprint("populations", __name__) + -iset = Blueprint("inbredset", __name__) +def create_resource( + cursor: db.DbCursor, + resource_name: str, + user: User, + group: Group, + public: bool +) -> Resource: + """Convenience function to create a resource of type 'inbredset-group'.""" + cursor.execute("SELECT * FROM resource_categories " + "WHERE resource_category_key='inbredset-group'") + category = cursor.fetchone() + if category: + return _create_resource(cursor, + resource_name, + ResourceCategory( + resource_category_id=uuid.UUID( + category["resource_category_id"]), + resource_category_key="inbredset-group", + resource_category_description=category[ + "resource_category_description"]), + user, + group, + public) + raise NotFoundError("Could not find a 'inbredset-group' resource category.") -@iset.route("/resource-id/<int:speciesid>/<int:inbredsetid>") + +@popbp.route("/populations/resource-id/<int:speciesid>/<int:inbredsetid>", + methods=["GET"]) def resource_id_by_inbredset_id(speciesid: int, inbredsetid: int) -> Response: """Retrieve the resource ID for resource attached to the inbredset.""" def __res_by_iset_id__(conn): @@ -20,7 +64,7 @@ def resource_id_by_inbredset_id(speciesid: int, inbredsetid: int) -> Response: (speciesid, inbredsetid)) return cursor.fetchone() - res = with_db_connection(__res_by_iset_id__) + res = db.with_db_connection(__res_by_iset_id__) if res: resp = jsonify({"status": "success", "resource-id": res["resource_id"]}) else: @@ -34,3 +78,76 @@ def resource_id_by_inbredset_id(speciesid: int, inbredsetid: int) -> Response: resp.status_code = 404 return resp + + +@popbp.route("/populations/create", methods=["POST"]) +@require_oauth("profile group resource") +def create_population_resource(): + """Create a resource of type 'inbredset-group'.""" + with (require_oauth.acquire("profile group resource") as _token, + db.connection(app.config["AUTH_DB"]) as conn, + db.cursor(conn) as cursor): + + def __check_form__(form, usergroup) -> Either: + """Check form for errors.""" + errors: tuple[str, ...] = tuple() + + species_id = form.get("species_id") + if not bool(species_id): + errors = errors + ("Missing `species_id` value.",) + + population_id = form.get("population_id") + if not bool(population_id): + errors = errors + ("Missing `population_id` value.",) + + population_name = form.get("population_name") + if not bool(population_name): + errors = errors + ("Missing `population_name` value.",) + + population_fullname = form.get("population_fullname") + if not bool(population_fullname): + errors = errors + ("Missing `population_fullname` value.",) + + if bool(errors): + error_messages = "\n\t - ".join(errors) + return Left({ + "error": "Invalid Request Data!", + "error_description": error_messages + }) + + return Right({"formdata": form, "group": usergroup}) + + def __default_group_if_none__(group) -> Either: + if group.is_nothing(): + return admin_group(conn) + return Right(group.value) + + return __default_group_if_none__( + user_group(conn, _token.user) + ).then( + lambda group: __check_form__(request_json(), group) + ).then( + lambda formdata: { + **formdata, + "resource": create_resource( + cursor, + f"Population — {formdata['formdata']['population_name']}", + _token.user, + formdata["group"], + formdata["formdata"].get("public", "on") == "on")} + ).then( + lambda resource: { + **resource, + "resource": assign_inbredset_group_owner_role( + cursor, resource["resource"], _token.user)} + ).then( + lambda resource: link_data_to_resource( + cursor, + resource["resource"].resource_id, + resource["formdata"]["species_id"], + resource["formdata"]["population_id"], + resource["formdata"]["population_name"], + resource["formdata"]["population_fullname"]) + ).either( + lambda error: (jsonify(error), 400), + jsonify) |