🔒 Raise error if migration is done by wrong user

lamindb_setup/core/_hub_crud.py

@@ -181,6 +181,21 @@ def select_db_user_by_instance(instance_id: str, client: Client):
     return data[0]
 
 
+def select_write_db_user_by_instance(instance_id: str, client: Client):
+    """Get db_user for which client has permission."""
+    data = (
+        client.table("db_user")
+        .select("*")
+        .eq("instance_id", instance_id)
+        .eq("name", "write")
+        .execute()
+        .data
+    )
+    if len(data) == 0:
+        return None
+    return data[0]
+
+
 def _delete_instance_record(instance_id: UUID, client: Client) -> None:
     if not isinstance(instance_id, UUID):
         instance_id = UUID(instance_id)

lamindb_setup/core/django.py

@@ -8,6 +8,10 @@
 from lamin_utils import logger
 from ._settings_store import current_instance_settings_file
 from ._settings_instance import InstanceSettings
+from ._hub_crud import select_write_db_user_by_instance
+from ._hub_client import call_with_fallback_auth
+from ._hub_utils import LaminDsnModel
+
 
 IS_RUN_FROM_IPYTHON = getattr(builtins, "__IPYTHON__", False)
 IS_SETUP = False
@@ -40,6 +44,7 @@ def setup_django(
     from django.core.management import call_command
 
     # configuration
+
     if not settings.configured:
         default_db = dj_database_url.config(
             default=isettings.db,
@@ -98,6 +103,7 @@ def setup_django(
         return None
 
     if deploy_migrations:
+        check_db_user(isettings)
         call_command("migrate", verbosity=2)
         isettings._update_cloud_sqlite_file(unlock_cloud_sqlite=False)
     elif init:
@@ -111,3 +117,29 @@ def setup_django(
 
     if isettings.keep_artifacts_local:
         isettings._search_local_root()
+
+
+def check_db_user(isettings: InstanceSettings):
+    """
+    Verify if the current database user has the necessary permissions to perform migrations.
+
+    This function checks if the current database user matches the authorized write user for the given instance.
+    If there's a mismatch or the write user information is not found, it raises a SystemExit with a detailed error message.
+
+    Args:
+        isettings (InstanceSettings): The instance settings containing database configuration.
+
+    Raises:
+        SystemExit: If the current user doesn't have the required permissions or if there's a configuration mismatch.
+    """
+    write_db_user = call_with_fallback_auth(
+        select_write_db_user_by_instance, instance_id=isettings._id
+    )
+    current_db_user = LaminDsnModel(db=isettings.db).db
+
+    if not write_db_user or current_db_user.user != write_db_user["db_user_name"]:
+        raise SystemExit(
+            "❌ The current db user is not authorized to perform migration. Please ensure the following:\n"
+            f"1. You have admin permissions for this instance.\n"
+            f"2. The database URL in your local settings is correct, it should ends with '_root'.\n"
+        )