Implement database_privileges.get_owner_oid_and_curr_role_db_priv RPC endpoint

db/roles/operations/select.py

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

db/sql/00_msar.sql

@@ -974,6 +974,30 @@ SELECT COALESCE(jsonb_agg(priv_cte.p), '[]'::jsonb) FROM priv_cte;
 $$ LANGUAGE SQL STABLE RETURNS NULL ON NULL INPUT;
 
 
+CREATE OR REPLACE FUNCTION msar.get_owner_oid_and_curr_role_db_priv(db_name text) RETURNS jsonb AS $$/*
+Given a database name, returns a json object with database owner oid and database privileges
+for the role executing the function.
+
+The returned JSON object has the form:
+  {
+    "owner_oid": <int>,
+    "current_role_db_priv" [<str>]
+  }
+*/
+SELECT jsonb_build_object(
+  'owner_oid', pgd.datdba,
+  'current_role_db_priv', array_remove(
+    ARRAY[
+      CASE WHEN has_database_privilege(pgd.oid, 'CREATE') THEN 'CREATE' END,
+      CASE WHEN has_database_privilege(pgd.oid, 'TEMPORARY') THEN 'TEMPORARY' END,
+      CASE WHEN has_database_privilege(pgd.oid, 'CONNECT') THEN 'CONNECT' END
+    ], NULL
+  )
+) FROM pg_catalog.pg_database AS pgd
+WHERE pgd.datname = db_name;
+$$ LANGUAGE SQL STABLE RETURNS NULL ON NULL INPUT;
+
+
 ----------------------------------------------------------------------------------------------------
 ----------------------------------------------------------------------------------------------------
 -- ROLE MANIPULATION FUNCTIONS

docs/docs/api/rpc.md

@@ -83,7 +83,9 @@ To use an RPC function:
     options:
       members:
       - list_direct
+      - get_owner_oid_and_curr_role_db_priv
       - DBPrivileges
+      - CurrentDBPrivileges
 
 ## Database Setup
 

mathesar/rpc/database_privileges.py

@@ -3,7 +3,7 @@
 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 db.roles.operations.select import list_db_priv, get_curr_role_db_priv
 from mathesar.rpc.utils import connect
 from mathesar.models.base import Database
 from mathesar.rpc.exceptions.handlers import handle_rpc_exceptions
@@ -28,6 +28,25 @@ def from_dict(cls, d):
         )
 
 
+class CurrentDBPrivileges(TypedDict):
+    """
+    Information about database privileges for current user.
+
+    Attributes:
+        owner_oid: The `oid` of the owner of the database.
+        current_role_db_priv: A list of database privileges for the current user.
+    """
+    owner_oid: int
+    current_role_db_priv: list[str]
+
+    @classmethod
+    def from_dict(cls, d):
+        return cls(
+            owner_oid=d["owner_oid"],
+            current_role_db_priv=d["current_role_db_priv"]
+        )
+
+
 @rpc_method(name="database_privileges.list_direct")
 @http_basic_auth_login_required
 @handle_rpc_exceptions
@@ -46,3 +65,24 @@ def list_direct(*, database_id: int, **kwargs) -> list[DBPrivileges]:
         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]
+
+
+# TODO: Think of something concise for the endpoint name.
+@rpc_method(name="database_privileges.get_owner_oid_and_curr_role_db_priv")
+@http_basic_auth_login_required
+@handle_rpc_exceptions
+def get_owner_oid_and_curr_role_db_priv(*, database_id: int, **kwargs) -> CurrentDBPrivileges:
+    """
+    Get database privileges for the current user.
+
+    Args:
+        database_id: The Django id of the database.
+
+    Returns:
+        A dict describing current user's database privilege.
+    """
+    user = kwargs.get(REQUEST_KEY).user
+    with connect(database_id, user) as conn:
+        db_name = Database.objects.get(id=database_id).name
+        curr_role_db_priv = get_curr_role_db_priv(db_name, conn)
+    return CurrentDBPrivileges.from_dict(curr_role_db_priv)

mathesar/tests/rpc/test_endpoints.py

@@ -135,6 +135,11 @@
         "database_privileges.list_direct",
         [user_is_authenticated]
     ),
+    (
+        database_privileges.get_owner_oid_and_curr_role_db_priv,
+        "database_privileges.get_owner_oid_and_curr_role_db_priv",
+        [user_is_authenticated]
+    ),
     (
         database_setup.create_new,
         "database_setup.create_new",