"""Handle the management of resources.""" import json from uuid import UUID, uuid4 from typing import Any, Dict, Sequence, NamedTuple from gn3.auth import db from gn3.auth.dictify import dictify from gn3.auth.authentication.users import User from .checks import authorised_for from ..checks import authorised_p from ..errors import NotFoundError, AuthorisationError from ..groups.models import Group, user_group, group_by_id, is_group_leader class MissingGroupError(AuthorisationError): """Raised for any resource operation without a group.""" class ResourceCategory(NamedTuple): """Class representing a resource category.""" resource_category_id: UUID resource_category_key: str resource_category_description: str def dictify(self) -> dict[str, Any]: """Return a dict representation of `ResourceCategory` objects.""" return { "resource_category_id": self.resource_category_id, "resource_category_key": self.resource_category_key, "resource_category_description": self.resource_category_description } class Resource(NamedTuple): """Class representing a resource.""" group: Group resource_id: UUID resource_name: str resource_category: ResourceCategory public: bool def dictify(self) -> dict[str, Any]: """Return a dict representation of `Resource` objects.""" return { "group": dictify(self.group), "resource_id": self.resource_id, "resource_name": self.resource_name, "resource_category": dictify(self.resource_category), "public": self.public } def __assign_resource_owner_role__(cursor, resource, user): """Assign `user` the 'Resource Owner' role for `resource`.""" cursor.execute( "SELECT gr.* FROM group_roles AS gr INNER JOIN roles AS r " "ON gr.role_id=r.role_id WHERE r.role_name='resource-owner'") role = cursor.fetchone() if not role: cursor.execute("SELECT * FROM roles WHERE role_name='resource-owner'") role = cursor.fetchone() cursor.execute( "INSERT INTO group_roles VALUES " "(:group_role_id, :group_id, :role_id)", {"group_role_id": str(uuid4()), "group_id": str(resource.group.group_id), "role_id": role["role_id"]}) cursor.execute( "INSERT INTO group_user_roles_on_resources " "VALUES (" ":group_id, :user_id, :role_id, :resource_id" ")", {"group_id": str(resource.group.group_id), "user_id": str(user.user_id), "role_id": role["role_id"], "resource_id": str(resource.resource_id)}) @authorised_p(("group:resource:create-resource",), error_description="Insufficient privileges to create a resource", oauth2_scope="profile resource") def create_resource( conn: db.DbConnection, resource_name: str, resource_category: ResourceCategory, user: User) -> Resource: """Create a resource item.""" with db.cursor(conn) as cursor: group = user_group(cursor, user).maybe( False, lambda grp: grp)# type: ignore[misc, arg-type] if not group: raise MissingGroupError( "User with no group cannot create a resource.") resource = Resource(group, uuid4(), resource_name, resource_category, False) cursor.execute( "INSERT INTO resources VALUES (?, ?, ?, ?, ?)", (str(resource.group.group_id), str(resource.resource_id), resource_name, str(resource.resource_category.resource_category_id), 1 if resource.public else 0)) __assign_resource_owner_role__(cursor, resource, user) return resource def resource_category_by_id( conn: db.DbConnection, category_id: UUID) -> ResourceCategory: """Retrieve a resource category by its ID.""" with db.cursor(conn) as cursor: cursor.execute( "SELECT * FROM resource_categories WHERE " "resource_category_id=?", (str(category_id),)) results = cursor.fetchone() if results: return ResourceCategory( UUID(results["resource_category_id"]), results["resource_category_key"], results["resource_category_description"]) raise NotFoundError( f"Could not find a ResourceCategory with ID '{category_id}'") def resource_categories(conn: db.DbConnection) -> Sequence[ResourceCategory]: """Retrieve all available resource categories""" with db.cursor(conn) as cursor: cursor.execute("SELECT * FROM resource_categories") return tuple( ResourceCategory(UUID(row[0]), row[1], row[2]) for row in cursor.fetchall()) return tuple() def public_resources(conn: db.DbConnection) -> Sequence[Resource]: """List all resources marked as public""" categories = { str(cat.resource_category_id): cat for cat in resource_categories(conn) } with db.cursor(conn) as cursor: cursor.execute("SELECT * FROM resources WHERE public=1") results = cursor.fetchall() group_uuids = tuple(row[0] for row in results) query = ("SELECT * FROM groups WHERE group_id IN " f"({', '.join(['?'] * len(group_uuids))})") cursor.execute(query, group_uuids) groups = { row[0]: Group( UUID(row[0]), row[1], json.loads(row[2] or "{}")) for row in cursor.fetchall() } return tuple( Resource(groups[row[0]], UUID(row[1]), row[2], categories[row[3]], bool(row[4])) for row in results) def group_leader_resources( cursor: db.DbCursor, user: User, group: Group, res_categories: Dict[UUID, ResourceCategory]) -> Sequence[Resource]: """Return all the resources available to the group leader""" if is_group_leader(cursor, user, group): cursor.execute("SELECT * FROM resources WHERE group_id=?", (str(group.group_id),)) return tuple( Resource(group, UUID(row[1]), row[2], res_categories[UUID(row[3])], bool(row[4])) for row in cursor.fetchall()) return tuple() def user_resources(conn: db.DbConnection, user: User) -> Sequence[Resource]: """List the resources available to the user""" categories = { # Repeated in `public_resources` function cat.resource_category_id: cat for cat in resource_categories(conn) } with db.cursor(conn) as cursor: def __all_resources__(group) -> Sequence[Resource]: gl_resources = group_leader_resources(cursor, user, group, categories) cursor.execute( ("SELECT resources.* FROM group_user_roles_on_resources " "LEFT JOIN resources " "ON group_user_roles_on_resources.resource_id=resources.resource_id " "WHERE group_user_roles_on_resources.group_id = ? " "AND group_user_roles_on_resources.user_id = ?"), (str(group.group_id), str(user.user_id))) rows = cursor.fetchall() private_res = tuple( Resource(group, UUID(row[1]), row[2], categories[UUID(row[3])], bool(row[4])) for row in rows) return tuple({ res.resource_id: res for res in (private_res + gl_resources + public_resources(conn))# type: ignore[operator] }.values()) # Fix the typing here return user_group(cursor, user).map(__all_resources__).maybe(# type: ignore[arg-type,misc] public_resources(conn), lambda res: res)# type: ignore[arg-type,return-value] def resource_by_id( conn: db.DbConnection, user: User, resource_id: UUID) -> Resource: """Retrieve a resource by its ID.""" if not authorised_for( conn, user, ("group:resource:view-resource",), (resource_id,))[resource_id]: raise AuthorisationError( "You are not authorised to access resource with id " f"'{resource_id}'.") with db.cursor(conn) as cursor: cursor.execute("SELECT * FROM resources WHERE resource_id=:id", {"id": str(resource_id)}) row = cursor.fetchone() if row: return Resource( group_by_id(conn, UUID(row["group_id"])), UUID(row["resource_id"]), row["resource_name"], resource_category_by_id(conn, row["resource_category_id"]), bool(int(row["public"]))) raise NotFoundError(f"Could not find a resource with id '{resource_id}'")