diff --git a/lib/init/grass.py b/lib/init/grass.py
index 87a69f5ef65..43d4989a303 100755
--- a/lib/init/grass.py
+++ b/lib/init/grass.py
@@ -1297,64 +1297,6 @@ def set_language(grass_config_dir):
     gettext.install("grasslibs", gpath("locale"))
 
 
-def lock_mapset(mapset_path, force_gislock_removal, user):
-    """Lock the mapset and return name of the lock file
-
-    Behavior on error must be changed somehow; now it fatals but GUI case is
-    unresolved.
-    """
-    if not os.path.exists(mapset_path):
-        fatal(_("Path '%s' doesn't exist") % mapset_path)
-    if not os.access(mapset_path, os.W_OK):
-        error = _("Path '%s' not accessible.") % mapset_path
-        stat_info = os.stat(mapset_path)
-        mapset_uid = stat_info.st_uid
-        if mapset_uid != os.getuid():
-            # GTC %s is mapset's folder path
-            error = "%s\n%s" % (
-                error,
-                _("You are not the owner of '%s'.") % mapset_path,
-            )
-        fatal(error)
-    # Check for concurrent use
-    lockfile = os.path.join(mapset_path, ".gislock")
-    ret = call([gpath("etc", "lock"), lockfile, "%d" % os.getpid()])
-    msg = None
-    if ret == 2:
-        if not force_gislock_removal:
-            msg = _(
-                "%(user)s is currently running GRASS in selected mapset"
-                " (file %(file)s found). Concurrent use not allowed.\n"
-                "You can force launching GRASS using -f flag"
-                " (note that you need permission for this operation)."
-                " Have another look in the processor "
-                "manager just to be sure..." % {"user": user, "file": lockfile}
-            )
-        else:
-            try_remove(lockfile)
-            message(
-                _(
-                    "%(user)s is currently running GRASS in selected mapset"
-                    " (file %(file)s found). Forcing to launch GRASS..."
-                    % {"user": user, "file": lockfile}
-                )
-            )
-    elif ret != 0:
-        msg = (
-            _("Unable to properly access '%s'.\nPlease notify system personnel.")
-            % lockfile
-        )
-
-    if msg:
-        raise Exception(msg)
-    debug(
-        "Mapset <{mapset}> locked using '{lockfile}'".format(
-            mapset=mapset_path, lockfile=lockfile
-        )
-    )
-    return lockfile
-
-
 # TODO: the gisrcrc here does not make sense, remove it from load_gisrc
 def unlock_gisrc_mapset(gisrc, gisrcrc):
     """Unlock mapset from the gisrc file"""
@@ -2436,14 +2378,16 @@ def main():
 
     location = mapset_settings.full_mapset
 
+    from grass.app.data import lock_mapset, MapsetLockingException
+
     try:
         # check and create .gislock file
         lock_mapset(
-            mapset_settings.full_mapset,
-            user=user,
-            force_gislock_removal=params.force_gislock_removal,
+            mapset_path=mapset_settings.full_mapset,
+            force_lock_removal=params.force_gislock_removal,
+            message_callback=message,
         )
-    except Exception as e:
+    except MapsetLockingException as e:
         fatal(e.args[0])
         sys.exit(_("Exiting..."))
 
diff --git a/python/grass/app/data.py b/python/grass/app/data.py
index 439a6c3c4d1..ec9ea9ff2f6 100644
--- a/python/grass/app/data.py
+++ b/python/grass/app/data.py
@@ -15,8 +15,12 @@
 import os
 import tempfile
 import getpass
+import subprocess
 import sys
 from shutil import copytree, ignore_patterns
+from pathlib import Path
+
+import grass.script as gs
 import grass.grassdb.config as cfg
 
 from grass.grassdb.checks import is_location_valid
@@ -162,3 +166,67 @@ def ensure_default_data_hierarchy():
     mapset_path = os.path.join(gisdbase, location, mapset)
 
     return gisdbase, location, mapset, mapset_path
+
+
+class MapsetLockingException(Exception):
+    pass
+
+
+def lock_mapset(mapset_path, force_lock_removal, message_callback):
+    """Acquire a lock for a mapset and return name of new lock file
+
+    Raises MapsetLockingException when it is not possible to acquire a lock for the
+    given mapset either becuase of existing lock or due to insufficient permissions.
+    A corresponding localized message is given in the exception.
+
+    A *message_callback* is a function which will be called to report messages about
+    certain states. Specifically, the function is called when forcibly unlocking the
+    mapset.
+
+    Assumes that the runtime is set up (GISBASE).
+    """
+    if not os.path.exists(mapset_path):
+        raise MapsetLockingException(_("Path '{}' doesn't exist").format(mapset_path))
+    if not os.access(mapset_path, os.W_OK):
+        error = _("Path '{}' not accessible.").format(mapset_path)
+        stat_info = os.stat(mapset_path)
+        mapset_uid = stat_info.st_uid
+        if mapset_uid != os.getuid():
+            error = "{error}\n{detail}".format(
+                error=error,
+                detail=_("You are not the owner of '{}'.").format(mapset_path),
+            )
+        raise MapsetLockingException(error)
+    # Check for concurrent use
+    lockfile = os.path.join(mapset_path, ".gislock")
+    locker_path = os.path.join(os.environ["GISBASE"], "etc", "lock")
+    ret = subprocess.run(
+        [locker_path, lockfile, "%d" % os.getpid()], check=False
+    ).returncode
+    msg = None
+    if ret == 2:
+        if not force_lock_removal:
+            msg = _(
+                "{user} is currently running GRASS in selected mapset"
+                " (file {file} found). Concurrent use of one mapset not allowed.\n"
+                "You can force launching GRASS using -f flag"
+                " (assuming your have sufficient access permissions)."
+                " Confirm in a process manager "
+                "that there is no other process using the mapset."
+            ).format(user=Path(lockfile).owner(), file=lockfile)
+        else:
+            message_callback(
+                _(
+                    "{user} is currently running GRASS in selected mapset"
+                    " (file {file} found), but forcing to launch GRASS anyway..."
+                ).format(user=Path(lockfile).owner(), file=lockfile)
+            )
+            gs.try_remove(lockfile)
+    elif ret != 0:
+        msg = _(
+            "Unable to properly access '{}'.\nPlease notify your system administrator."
+        ).format(lockfile)
+
+    if msg:
+        raise MapsetLockingException(msg)
+    return lockfile