auth: Enable retrieval of a role by id.

2 files changed, 42 insertions, 3 deletions

diff --git a/gn3/auth/authorisation/roles.py b/gn3/auth/authorisation/roles.py
index 86759b1..e75163d 100644
--- a/gn3/auth/authorisation/roles.py
+++ b/gn3/auth/authorisation/roles.py
@@ -4,6 +4,7 @@ from functools import reduce
 from typing import Any, Sequence, Iterable, NamedTuple
 
 from pymonad.maybe import Just, Maybe, Nothing
+from pymonad.either import Left, Right, Either
 
 from gn3.auth import db
 from gn3.auth.dictify import dictify
@@ -12,6 +13,7 @@ from gn3.auth.authentication.checks import authenticated_p
 
 from .checks import authorised_p
 from .privileges import Privilege
+from .errors import AuthorisationError
 
 class Role(NamedTuple):
     """Class representing a role: creates immutable objects."""
@@ -25,6 +27,11 @@ class Role(NamedTuple):
             "role_id": self.role_id, "role_name": self.role_name,
             "privileges": tuple(dictify(priv) for priv in self.privileges)
         }
+
+class RoleNotFoundError(AuthorisationError):
+    """Raised whenever we try fetching (a) role(s) that do(es) not exist."""
+    error_code: int = 404
+
 @authenticated_p
 @authorised_p(("group:role:create-role",), error_message="Could not create role")
 def create_role(
@@ -93,6 +100,24 @@ def user_roles(conn: db.DbConnection, user: User) -> Maybe[Sequence[Role]]:
                 reduce(__organise_privileges__, results, {}).values()))
         return Nothing
 
+def user_role(conn: db.DbConnection, user: User, role_id: UUID) -> Either:
+    """Retrieve a specific non-resource role assigned to the user."""
+    with db.cursor(conn) as cursor:
+        cursor.execute(
+            "SELECT r.*, p.* FROM user_roles AS ur INNER JOIN roles AS r "
+            "ON ur.role_id=r.role_id INNER JOIN role_privileges AS rp "
+            "ON r.role_id=rp.role_id INNER JOIN privileges AS p "
+            "ON rp.privilege_id=p.privilege_id "
+            "WHERE ur.user_id=? AND ur.role_id=?",
+            (str(user.user_id), str(role_id)))
+
+        results = cursor.fetchall()
+        if results:
+            return Right(tuple(
+                reduce(__organise_privileges__, results, {}).values())[0])
+        return Left(RoleNotFoundError(
+            f"Could not find role with id '{role_id}'",))
+
 def assign_default_roles(cursor: db.DbCursor, user: User):
     """Assign `user` some default roles."""
     cursor.execute(
diff --git a/gn3/auth/authorisation/views.py b/gn3/auth/authorisation/views.py
index 11e43eb..6c6e5e3 100644
--- a/gn3/auth/authorisation/views.py
+++ b/gn3/auth/authorisation/views.py
@@ -1,16 +1,17 @@
 """Endpoints for the authorisation stuff."""
+import uuid
 import traceback
 from typing import Tuple, Optional
 
 import sqlite3
-from flask import request, jsonify, current_app
+from flask import request, jsonify, Response, current_app
 
 from gn3.auth import db
 from gn3.auth.dictify import dictify
 from gn3.auth.blueprint import oauth2
 
 from .errors import UserRegistrationError
-from .roles import assign_default_roles, user_roles as _user_roles
+from .roles import user_role, assign_default_roles, user_roles as _user_roles
 from .groups import (
     user_group, all_groups, GroupCreationError, create_group as _create_group)
 
@@ -41,7 +42,7 @@ def user_roles():
     with require_oauth.acquire("role") as token:
         with db.connection(current_app.config["AUTH_DB"]) as conn:
             return jsonify(_user_roles(conn, token.user).maybe(
-                tuple(), lambda rls: rls))
+                tuple(), lambda roles: tuple(dictify(role) for role in roles)))
 
 def __email_valid__(email: str) -> Tuple[bool, Optional[str]]:
     """Validate the email address."""
@@ -145,3 +146,16 @@ def create_group():
             return jsonify({
                 **dictify(new_group), "group_leader": dictify(user)
             })
+
+@oauth2.route("/role/<uuid:role_id>", methods=["GET"])
+@require_oauth("role")
+def role(role_id: uuid.UUID) -> Response:
+    """Retrieve a user role with id `role_id`"""
+    def __error__(exc: Exception):
+        raise exc
+    with require_oauth.acquire("profile role") as the_token:
+        db_uri = current_app.config["AUTH_DB"]
+        with db.connection(db_uri) as conn:
+            the_role = user_role(conn, the_token.user, role_id)
+            return the_role.either(
+                __error__, lambda a_role: jsonify(dictify(a_role)))