about summary refs log tree commit diff
path: root/gn_auth/auth/authorisation/roles
diff options
context:
space:
mode:
Diffstat (limited to 'gn_auth/auth/authorisation/roles')
-rw-r--r--gn_auth/auth/authorisation/roles/models.py23
-rw-r--r--gn_auth/auth/authorisation/roles/views.py10
2 files changed, 24 insertions, 9 deletions
diff --git a/gn_auth/auth/authorisation/roles/models.py b/gn_auth/auth/authorisation/roles/models.py
index 2729b3b..89556a6 100644
--- a/gn_auth/auth/authorisation/roles/models.py
+++ b/gn_auth/auth/authorisation/roles/models.py
@@ -4,12 +4,12 @@ from functools import reduce
 from dataclasses import dataclass
 from typing import Sequence, Iterable, Optional
 
+from gn_libs import sqlite3 as db
 from pymonad.either import Left, Right, Either
 
 from gn_auth.auth.errors import NotFoundError, AuthorisationError
 from gn_auth.auth.authorisation.resources.base import Resource
 
-from ...db import sqlite3 as db
 from ...authentication.users import User
 
 from ..checks import authorised_p
@@ -255,6 +255,25 @@ def assign_user_role_by_name(
             })
 
 
+def unassign_user_role_by_name(
+        cursor: db.DbCursor, user: User, resource_id: UUID, role_name: str):
+    """Revoke a role from `user` on `resource_id` by the role's name."""
+    cursor.execute(
+        "SELECT role_id FROM roles WHERE role_name=:role_name",
+        {"role_name": role_name})
+    role = cursor.fetchone()
+    if role:
+        cursor.execute(
+            ("DELETE FROM user_roles "
+             "WHERE user_id=:user_id AND role_id=:role_id "
+             "AND resource_id=:resource_id"),
+            {
+                "user_id": str(user.user_id),
+                "role_id": role["role_id"],
+                "resource_id": str(resource_id)
+            })
+
+
 def role_by_id(conn: db.DbConnection, role_id: UUID) -> Optional[Role]:
     """Fetch a role from the database by its ID."""
     with db.cursor(conn) as cursor:
@@ -271,7 +290,7 @@ def role_by_id(conn: db.DbConnection, role_id: UUID) -> Optional[Role]:
 
     _roles = db_rows_to_roles(results)
     if len(_roles) > 1:
-        raise Exception("Data corruption: Expected a single role.")
+        raise Exception("Data corruption: Expected a single role.")# pylint: disable=[broad-exception-raised]
 
     return _roles[0]
 
diff --git a/gn_auth/auth/authorisation/roles/views.py b/gn_auth/auth/authorisation/roles/views.py
index 00def89..91292e7 100644
--- a/gn_auth/auth/authorisation/roles/views.py
+++ b/gn_auth/auth/authorisation/roles/views.py
@@ -7,7 +7,7 @@ from flask import jsonify, Response, Blueprint, current_app
 
 from ...db import sqlite3 as db
 
-from .models import user_role
+from .models import role_by_id
 
 from ...authentication.oauth2.resource_server import require_oauth
 
@@ -17,11 +17,7 @@ roles = Blueprint("roles", __name__)
 @require_oauth("profile role")
 def view_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:
+    with require_oauth.acquire("profile role") as _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((asdict(a_role[0]), str(a_role[1]))))
+            return jsonify(asdict(role_by_id(conn, role_id)))# type: ignore[arg-type]