block updates to read only registry tables

latch/registry/table.py

@@ -1,3 +1,4 @@
+import json
 import sys
 import time
 from contextlib import contextmanager
@@ -76,6 +77,7 @@ class _ColumnNode(TypedDict("_ColumnNodeReserved", {"def": DBValue})):
 
 @dataclass
 class _Cache:
+    read_only: Optional[bool] = None
     display_name: Optional[str] = None
     columns: Optional[Dict[str, Column]] = None
     project_id: Optional[str] = None
@@ -123,6 +125,7 @@ def load(self) -> None:
                             }
                         }
                         projectId
+                        metadata
                     }
                 }
                 """),
@@ -134,6 +137,13 @@ def load(self) -> None:
                 f"table does not exist or you lack permissions: id={self.id}"
             )
 
+        try:
+            metadata = data["metadata"]
+            read_only = "benchlingSchemaId" in json.loads(metadata)
+        except (KeyError, TypeError, json.JSONDecodeError):
+            read_only = False
+
+        self._cache.read_only = read_only
         self._cache.project_id = data["projectId"]
         self._cache.display_name = data["displayName"]
 
@@ -233,6 +243,28 @@ def get_columns(
 
         return self._cache.columns
 
+    @overload
+    def read_only(self, *, load_if_missing: Literal[True] = True) -> bool: ...
+
+    @overload
+    def read_only(self, *, load_if_missing: bool) -> Optional[bool]: ...
+
+    def read_only(self, *, load_if_missing: bool = True) -> Optional[bool]:
+        """Check whether or not this table is read-only
+
+        Args:
+            load_if_missing:
+                If true, :meth:`load` the read_only status if not in cache.
+                If false, return `None` if not in cache.
+
+        Returns:
+            Mapping between column keys and :class:`columns <latch.registry.types.Column>`.
+        """
+        if self._cache.read_only is None and load_if_missing:
+            self.load()
+
+        return self._cache.read_only
+
     def list_records(self, *, page_size: int = 100) -> Iterator[Dict[str, Record]]:
         """List Registry records contained in this table.
 
@@ -364,6 +396,9 @@ def update(self, *, reload_on_commit: bool = True) -> Iterator["TableUpdate"]:
             Context manager for the new transaction.
         """
 
+        if self.read_only():
+            raise RuntimeError(f"cannot update read-only table: id={self.id}")
+
         upd = TableUpdate(self)
         yield upd
         upd.commit()