"""Utilities common to more than one resource.""" import uuid from gn_auth.auth.db import sqlite3 as db def assign_resource_owner_role( cursor: db.DbCursor, resource_id: uuid.UUID, user_id: uuid.UUID ) -> dict: """Assign `user` the 'Resource Owner' role for `resource`.""" cursor.execute("SELECT * FROM roles WHERE role_name='resource-owner'") role = cursor.fetchone() params = { "user_id": str(user_id), "role_id": role["role_id"], "resource_id": str(resource_id) } cursor.execute( "INSERT INTO user_roles " "VALUES (:user_id, :role_id, :resource_id) " "ON CONFLICT (user_id, role_id, resource_id) DO NOTHING", params) return params def grant_access_to_sysadmins( cursor: db.DbCursor, resource_id: uuid.UUID, system_resource_id: uuid.UUID ): """Grant sysadmins access to resource identified by `resource_id`.""" cursor.execute( "SELECT role_id FROM roles WHERE role_name='system-administrator'") sysadminroleid = cursor.fetchone()[0] cursor.execute(# Fetch sysadmin IDs. "SELECT user_roles.user_id FROM roles INNER JOIN user_roles " "ON roles.role_id=user_roles.role_id " "WHERE role_name='system-administrator' AND resource_id=?", (str(system_resource_id),)) cursor.executemany( "INSERT INTO user_roles(user_id, role_id, resource_id) " "VALUES (?, ?, ?)", tuple((row["user_id"], sysadminroleid, str(resource_id)) for row in cursor.fetchall()))