aboutsummaryrefslogtreecommitdiff
path: root/gn_auth/auth/authorisation/resources
diff options
context:
space:
mode:
Diffstat (limited to 'gn_auth/auth/authorisation/resources')
-rw-r--r--gn_auth/auth/authorisation/resources/inbredset/models.py96
-rw-r--r--gn_auth/auth/authorisation/resources/inbredset/views.py77
2 files changed, 172 insertions, 1 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..de1c18a
--- /dev/null
+++ b/gn_auth/auth/authorisation/resources/inbredset/models.py
@@ -0,0 +1,96 @@
+"""Functions to handle the low-level details regarding populations auth."""
+from uuid import UUID, uuid4
+
+import sqlite3
+
+from gn_auth.auth.errors import NotFoundError
+from gn_auth.auth.authentication.users import User
+from gn_auth.auth.authorisation.resources.groups.models import Group
+from gn_auth.auth.authorisation.resources.base import Resource, ResourceCategory
+from gn_auth.auth.authorisation.resources.models import (
+ create_resource as _create_resource)
+
+def create_resource(
+ cursor: sqlite3.Cursor,
+ 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(
+ 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.")
+
+
+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]
+ 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
diff --git a/gn_auth/auth/authorisation/resources/inbredset/views.py b/gn_auth/auth/authorisation/resources/inbredset/views.py
index d565533..b559105 100644
--- a/gn_auth/auth/authorisation/resources/inbredset/views.py
+++ b/gn_auth/auth/authorisation/resources/inbredset/views.py
@@ -1,8 +1,17 @@
"""Views for InbredSet resources."""
-from flask import jsonify, Response, Blueprint
+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.requests import request_json
from gn_auth.auth.db.sqlite3 import with_db_connection
+from gn_auth.auth.authentication.oauth2.resource_server import require_oauth
+from gn_auth.auth.authorisation.resources.groups.models import user_group
+
+from .models import (create_resource,
+ link_data_to_resource,
+ assign_inbredset_group_owner_role)
popbp = Blueprint("populations", __name__)
@@ -35,3 +44,69 @@ 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})
+
+ return 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)