Implement database_privileges.list_direct RPC endpoint.

config/settings/common_settings.py

@@ -71,6 +71,7 @@ def pipe_delim(pipe_string):
     'mathesar.rpc.constraints',
     'mathesar.rpc.columns',
     'mathesar.rpc.columns.metadata',
+    'mathesar.rpc.database_privileges',
     'mathesar.rpc.database_setup',
     'mathesar.rpc.databases',
     'mathesar.rpc.records',

db/roles/operations/select.py

@@ -3,3 +3,7 @@
 
 def get_roles(conn):
     return exec_msar_func(conn, 'get_roles').fetchone()[0]
+
+
+def list_db_priv(db_name, conn):
+    return exec_msar_func(conn, 'list_db_priv', db_name).fetchone()[0]

db/sql/00_msar.sql

@@ -894,7 +894,7 @@ FROM (
     s.oid,
     s.nspname
 ) AS schema_data;
-$$ LANGUAGE sql;
+$$ LANGUAGE SQL;
 
 
 CREATE OR REPLACE FUNCTION msar.get_roles() RETURNS jsonb AS $$/*
@@ -945,7 +945,33 @@ FROM (
     LEFT OUTER JOIN rolemembers ON r.oid = rolemembers.oid
   WHERE r.rolname NOT LIKE 'pg_%'
 ) AS role_data;
-$$ LANGUAGE sql;
+$$ LANGUAGE SQL STABLE;
+
+
+CREATE OR REPLACE FUNCTION msar.list_db_priv(db_name text) RETURNS jsonb AS $$/*
+Given a database name, returns a json array of objects with database privileges for non-inherited roles.
+
+Each returned JSON object in the array has the form:
+  {
+    "role_oid": <int>,
+    "direct" [<str>]
+  }
+*/
+WITH priv_cte AS (
+  SELECT
+    jsonb_build_object(
+      'role_oid', pgr.oid,
+      'direct',  jsonb_agg(acl.privilege_type)
+    ) AS p
+  FROM
+    pg_catalog.pg_roles AS pgr,
+    pg_catalog.pg_database AS pgd,
+    aclexplode(COALESCE(pgd.datacl, acldefault('d', pgd.datdba))) AS acl
+  WHERE pgd.datname = db_name AND pgr.oid = acl.grantee AND pgr.rolname NOT LIKE 'pg_'
+  GROUP BY pgr.oid, pgd.oid
+)
+SELECT COALESCE(jsonb_agg(priv_cte.p), '[]'::jsonb) FROM priv_cte;
+$$ LANGUAGE SQL STABLE RETURNS NULL ON NULL INPUT;
 
 
 ----------------------------------------------------------------------------------------------------

docs/docs/api/rpc.md

@@ -77,6 +77,14 @@ To use an RPC function:
       - list_
       - DatabaseInfo
 
+## Database Privileges
+
+::: database_privileges
+    options:
+      members:
+      - list_direct
+      - DBPrivileges
+
 ## Database Setup
 
 ::: database_setup

mathesar/rpc/database_privileges.py

@@ -0,0 +1,48 @@
+from typing import TypedDict
+
+from modernrpc.core import rpc_method, REQUEST_KEY
+from modernrpc.auth.basic import http_basic_auth_login_required
+
+from db.roles.operations.select import list_db_priv
+from mathesar.rpc.utils import connect
+from mathesar.models.base import Database
+from mathesar.rpc.exceptions.handlers import handle_rpc_exceptions
+
+
+class DBPrivileges(TypedDict):
+    """
+    Information about database privileges.
+
+    Attributes:
+        role_oid: The `oid` of the role on the database server.
+        direct: A list of database privileges for the afforementioned role_oid.
+    """
+    role_oid: int
+    direct: list[str]
+
+    @classmethod
+    def from_dict(cls, d):
+        return cls(
+            role_oid=d["role_oid"],
+            direct=d["direct"]
+        )
+
+
+@rpc_method(name="database_privileges.list_direct")
+@http_basic_auth_login_required
+@handle_rpc_exceptions
+def list_direct(*, database_id: int, **kwargs) -> list[DBPrivileges]:
+    """
+    List database privileges for non-inherited roles.
+
+    Args:
+        database_id: The Django id of the database.
+
+    Returns:
+        A list of database privileges.
+    """
+    user = kwargs.get(REQUEST_KEY).user
+    with connect(database_id, user) as conn:
+        db_name = Database.objects.get(id=database_id).name
+        raw_db_priv = list_db_priv(db_name, conn)
+    return [DBPrivileges.from_dict(i) for i in raw_db_priv]

mathesar/tests/rpc/test_endpoints.py

@@ -13,6 +13,7 @@
 from mathesar.rpc import configured_roles
 from mathesar.rpc import connections
 from mathesar.rpc import constraints
+from mathesar.rpc import database_privileges
 from mathesar.rpc import database_setup
 from mathesar.rpc import databases
 from mathesar.rpc import explorations
@@ -129,6 +130,11 @@
         "constraints.delete",
         [user_is_authenticated]
     ),
+    (
+        database_privileges.list_direct,
+        "database_privileges.list_direct",
+        [user_is_authenticated]
+    ),
     (
         database_setup.create_new,
         "database_setup.create_new",