Load predefined users from a JSON file through command line. #9229

Author: khushboovashi

docs/en_US/user_management.rst

@@ -270,6 +270,50 @@ username/email address.
     /path/to/python /path/to/setup.py get-users --username user1@gmail.com
 
 
+Load Users
+**********
+
+To bulk import users from a JSON file, invoke ``setup.py`` with ``load-users`` command line option,
+followed by the path to the JSON file.
+
+.. code-block:: bash
+
+    /path/to/python /path/to/setup.py load-users /path/to/users.json
+
+**JSON File Format**
+
+The input JSON file must contain a ``users`` array with user objects:
+
+.. code-block:: json
+
+    {
+        "users": [
+            {
+                "username": "admin@example.com",
+                "email": "admin@example.com",
+                "password": "securepassword",
+                "role": "Administrator",
+                "active": true,
+                "auth_source": "internal"
+            },
+            {
+                "username": "ldap_user",
+                "email": "ldap_user@example.com",
+                "role": "User",
+                "active": true,
+                "auth_source": "ldap"
+            }
+        ]
+    }
+
+The command handles errors gracefully:
+
+* Users that already exist are skipped
+* Invalid roles are reported and skipped
+* Missing passwords for internal auth are reported and skipped
+* Passwords shorter than 6 characters are reported and skipped
+
+
 Output
 ******
 

web/pgadmin/utils/session.py

@@ -26,7 +26,7 @@
 import config
 from uuid import uuid4
 from threading import Lock
-from flask import current_app, request, flash, redirect
+from flask import current_app, request, flash, redirect, has_request_context
 from flask_login import login_url
 
 from pickle import dump, load
@@ -115,6 +115,17 @@ def _normalize(self):
                 while len(self._cache) > (self.num_to_store * 0.8):
                     self._cache.popitem(False)
 
+    def is_session_ready(self, _session):
+        if not has_request_context() and _session is None:
+            return False
+
+        # Session _id returns the str object
+        # or None if it hasn't been set yet.
+        try:
+            return _session['_id'] is not None
+        except (AssertionError, RuntimeError, KeyError):
+            return False
+
     def new_session(self):
         session = self.parent.new_session()
 
@@ -143,16 +154,17 @@ def exists(self, sid):
 
     def get(self, sid, digest):
         session = None
-        with sess_lock:
+        with (sess_lock):
             if sid in self._cache:
                 session = self._cache[sid]
-                if session and session.hmac_digest != digest:
+                if self.is_session_ready(session) and\
+                        session.hmac_digest != digest:
                     session = None
 
                 # reset order in Dict
                 del self._cache[sid]
 
-            if not session:
+            if not self.is_session_ready(session):
                 session = self.parent.get(sid, digest)
 
             # Do not store the session if skip paths

web/setup.py

@@ -163,6 +163,159 @@ def get_role(role: str):
 
 class ManageUsers:
 
+    @app.command()
+    @update_sqlite_path
+    def load_users(input_file: str,
+                   sqlite_path: Optional[str] = None):
+        """Load users from a JSON file.
+
+        Expected JSON format:
+        {
+            "users": [
+                {
+                    "username": "user@example.com",
+                    "email": "user@example.com",
+                    "password": "password123",
+                    "role": "User",
+                    "active": true,
+                    "auth_source": "internal"
+                },
+                {
+                    "username": "ldap_user",
+                    "email": "ldap@example.com",
+                    "role": "Administrator",
+                    "active": true,
+                    "auth_source": "ldap"
+                }
+            ]
+        }
+        """
+        from urllib.parse import unquote
+
+        print('----------')
+        print('Loading users from:', input_file)
+        print('SQLite pgAdmin config:', config.SQLITE_PATH)
+        print('----------')
+
+        # Parse the input file path
+        try:
+            file_path = unquote(input_file)
+        except Exception as e:
+            return _handle_error(str(e), True)
+
+        # Read and parse JSON file
+        try:
+            with open(file_path) as f:
+                data = jsonlib.load(f)
+        except jsonlib.decoder.JSONDecodeError as e:
+            return _handle_error(
+                gettext("Error parsing input file %s: %s" % (file_path, e)),
+                True)
+        except Exception as e:
+            return _handle_error(
+                gettext("Error reading input file %s: [%d] %s" %
+                        (file_path, e.errno, e.strerror)), True)
+
+        # Validate JSON structure
+        if 'users' not in data:
+            return _handle_error(
+                gettext("Invalid JSON format: 'users' key not found"), True)
+
+        users_data = data['users']
+        if not isinstance(users_data, list):
+            return _handle_error(
+                gettext("Invalid JSON format: 'users' must be a list"), True)
+
+        created_count = 0
+        skipped_count = 0
+        error_count = 0
+
+        app = create_app(config.APP_NAME + '-cli')
+        with (app.test_request_context()):
+            for user_entry in users_data:
+                try:
+                    # Validate required fields
+                    if 'username' not in user_entry and\
+                            'email' not in user_entry:
+                        print(f"Skipping user: missing 'username' or 'email'")
+                        error_count += 1
+                        continue
+
+                    # Determine auth_source (default to internal)
+                    auth_source = user_entry.get('auth_source', INTERNAL)
+
+                    # Build user data dict
+                    user_data = {
+                        'username': user_entry.get('username',
+                                                   user_entry.get('email')),
+                        'email': user_entry.get('email'),
+                        'role': user_entry.get('role', 'User'),
+                        'active': user_entry.get('active', True),
+                        'auth_source': auth_source
+                    }
+
+                    # For internal auth, password is required
+                    if auth_source == INTERNAL:
+                        if 'password' not in user_entry:
+                            print(f"Skipping user '{user_data['username']}': "
+                                  f"password required for internal auth")
+                            error_count += 1
+                            continue
+                        user_data['newPassword'] = user_entry['password']
+                        user_data['confirmPassword'] = user_entry['password']
+
+                    # Check if user already exists
+                    usr = User.query.filter_by(username=user_data['username'],
+                                               auth_source=auth_source).first()
+
+                    uid = usr.id if usr else None
+
+                    if uid:
+                        print(f"Skipping user '{user_data['username']}': "
+                              f"already exists")
+                        skipped_count += 1
+                        continue
+
+                    # Get role ID
+                    role = Role.query.filter_by(name=user_data['role']).first()
+                    rid = role.id if role else None
+
+                    if rid is None:
+                        print(f"Skipping user '{user_data['username']}': "
+                              f"role '{user_data['role']}' does not exist")
+                        error_count += 1
+                        continue
+
+                    user_data['role'] = rid
+
+                    # Validate password length for internal users
+                    if auth_source == INTERNAL:
+                        if len(user_data['newPassword']) < 6:
+                            print(f"Skipping user '{user_data['username']}': "
+                                  f"password must be at least 6 characters")
+                            error_count += 1
+                            continue
+
+                    # Create the user
+                    status, msg = create_user(user_data)
+                    if status:
+                        print(f"Created user: {user_data['username']}")
+                        created_count += 1
+                    else:
+                        print(f"Error creating user '{user_data['username']}'"
+                              f": {msg}")
+                        error_count += 1
+
+                except Exception as e:
+                    print(f"Error processing user entry: {str(e)}")
+                    error_count += 1
+
+        print('----------')
+        print(f"Users created: {created_count}")
+        print(f"Users skipped (already exist): {skipped_count}")
+        print(f"Errors: {error_count}")
+        print('----------')
+
     @app.command()
     @update_sqlite_path
     def add_user(email: str, password: str,