diff --git a/sdk/basyx/aas/backend/local_file.py b/sdk/basyx/aas/backend/local_file.py
index df21ddfe..c00bb8da 100644
--- a/sdk/basyx/aas/backend/local_file.py
+++ b/sdk/basyx/aas/backend/local_file.py
@@ -11,7 +11,8 @@
 The :class:`~LocalFileIdentifiableStore` handles adding, deleting and otherwise managing
 the AAS objects in a specific Directory.
 """
-from typing import Iterator
+import contextlib
+from typing import Iterator, Generator, TextIO, Optional
 import logging
 import json
 import os
@@ -20,6 +21,7 @@
 import threading
 import warnings
 import weakref
+from pathlib import Path
 
 from ..adapter.json import json_serialization, json_deserialization
 from basyx.aas import model
@@ -28,23 +30,139 @@
 logger = logging.getLogger(__name__)
 
 
+class DirectoryLock:
+    """
+    Implementation of a Lock on a directory. Uses POSIX ``flock`` on ``<directory>/.lock`` to control which instance
+    holds the lock. Even if the process which holds the lock dies, the lock is released.
+
+    The methods :meth:`acquire` and :meth:`release` are thread-safe and can be called in any state. If :meth:`acquire`
+    is called when the directory is already locked no errors occur. The same holds for calling :meth:`release` in an
+    unlocked state.
+
+    .. note::
+        Directory locking relies on ``fcntl.flock``, which is only available on POSIX platforms. On Windows
+        a warning is logged and the lock is skipped.
+    """
+    def __init__(self, directory_path: str):
+        self.directory_path = Path(directory_path)
+        self.lock_path = self.directory_path / ".lock"
+        self._dir_lock_file: Optional[TextIO] = None
+
+        # We count the number of active accesses to the directory to ensure that
+        # release() is only releasing if all are done
+        self._locking_lock = threading.Condition()
+        self._is_locked_flag = False
+        self._active_accesses = 0
+        self._is_releasing = False
+
+    def _lazy_import_fcntl(self):
+        try:
+            import fcntl as _fcntl
+        except ImportError:
+            _fcntl = None  # Windows: directory locking is unavailable
+        return _fcntl
+
+    def acquire(self) -> None:
+        """
+        Acquires the lock on the directory non-blocking. If the ``<directory>/.lock`` file is already locked
+        it raises a :class:`RuntimeError`.
+
+        .. note::
+            Directory locking relies on ``fcntl.flock``, which is only available on POSIX platforms. On Windows
+            a warning is logged and the lock is skipped.
+
+        :raises RuntimeError: If the directory is already locked by another instance
+        """
+        with self._locking_lock:  # Ensure only one dir_lock is acquired at a time
+            if self._is_locked_flag:
+                return  # Idempotent
+
+            self._dir_lock_file = open(self.lock_path, "a")  # use "a" to not swap the inode as in "w"
+
+            _fcntl = self._lazy_import_fcntl()
+            if _fcntl is None:
+                # Windows does not support locking files. For now, we raise a warning
+                # and continue to assure backwards compatibility
+
+                logger.warning("fcntl unavailable; directory locking is disabled (Windows?)")
+                self._is_locked_flag = True
+                return
+
+            try:
+                _fcntl.flock(self._dir_lock_file.fileno(), _fcntl.LOCK_EX | _fcntl.LOCK_NB)
+            except OSError:
+                # dir_lock already taken by other process
+                self._dir_lock_file.close()
+                self._dir_lock_file = None
+                raise RuntimeError(f"Directory {self.directory_path} is already locked by another DirectoryLock")
+            else:
+                self._is_locked_flag = True
+
+    def release(self) -> None:
+        """
+        If the directory is currently locked, this method waits for all current accesses in :meth:`ensure_locked`
+        to finish and releases the lock afterwards.
+        """
+        with self._locking_lock:
+            if not self._is_locked_flag:
+                return  # Idempotent
+            self._is_releasing = True  # Restrict new ensure_locked() entries
+            while self._active_accesses > 0:  # Wait for all active accesses to finish
+                self._locking_lock.wait()
+
+            # Release flock
+            self._is_locked_flag = False
+            self._is_releasing = False
+            if self._dir_lock_file is not None:
+                self._dir_lock_file.close()  # lock is released by closing fd
+                self._dir_lock_file = None
+
+    @contextlib.contextmanager
+    def ensure_locked(self) -> Generator:
+        """Context Manager for access to the locked directory
+
+        Use this as a context manager when accessing files in the locked directory. This ensures that the directory
+        is locked during the whole execution of the ``with`` block. Fails with a :class:`RuntimeError` if the directory
+        is not locked when entering the ``with`` block.
+
+        :raises RuntimeError: If the directory is not locked on enter
+        """
+        with self._locking_lock:
+            if (not self._is_locked_flag) or self._is_releasing:
+                raise RuntimeError(f"Directory {self.directory_path} is not locked!")
+            self._active_accesses += 1
+        try:
+            yield
+        finally:
+            with self._locking_lock:
+                self._active_accesses -= 1
+                self._locking_lock.notify_all()
+
+    def close(self) -> None:
+        self.release()
+
+
 class LocalFileIdentifiableStore(model.AbstractObjectStore[model.Identifier, model.Identifiable]):
     """
     An ObjectStore implementation for :class:`~basyx.aas.model.base.Identifiable` BaSyx Python SDK objects backed
-    by a local file based local backend
-
-    .. warning::
-        This backend is intended for development and testing only. It provides no
-        concurrency control across processes: concurrent writes to the same object
-        (e.g. under a multi-worker WSGI server) will silently overwrite each other,
-        with the last writer winning and no error raised. Use a dedicated database
-        backend for any production deployment.
+    by a local file based local backend.
+
+    At most one instance of this class can manage a directory. To ensure this a
+    :class:`~basyx.aas.backend.local_file.DirectoryLock` is used (with all it limitations on non-POSIX systems).
+    The lock is applied eager (as soon as the directory exists) either in the constructor or after calling
+    :meth:`check_directory`. Unless this is done all further method calls will fail with :class:`RuntimeError`.
+
+    Call :meth:`close` to release the directory lock when done.
+
+    .. note::
+        Directory locking relies on ``fcntl.flock``, which is only available on POSIX platforms. On Windows
+        a warning is logged and the lock is skipped.
     """
     def __init__(self, directory_path: str):
         """
         Initializer of class LocalFileIdentifiableStore
 
-        :param directory_path: Path to the local file backend (the path where you want to store your AAS JSON files)
+        :param directory_path: Path to the local file backend (the path where you want to store your AAS JSON files).
         """
         self.directory_path: str = directory_path.rstrip("/")
 
@@ -57,6 +175,26 @@ def __init__(self, directory_path: str):
             = weakref.WeakValueDictionary()
         self._object_cache_lock = threading.Lock()
 
+        # Prevent concurrent write operations to avoid TOCTOU problems
+        self._write_lock = threading.Lock()
+
+        # We need to prevent multiple instances of LocalFileIdentifiableStore performing R/W operations on the same
+        # directory in order to ensure cache validity. The directory is locked as soon as it exists.
+        self._dir_lock = DirectoryLock(self.directory_path)
+        if os.path.exists(self.directory_path):
+            self._acquire_dir_lock()
+
+    def _acquire_dir_lock(self) -> None:
+        try:
+            self._dir_lock.acquire()
+        except RuntimeError as ex:
+            raise RuntimeError(f"Directory {self.directory_path} is already in use"
+                               f" by another LocalFileIdentifiableStore instance.") from ex
+
+    def close(self) -> None:
+        """Release the directory lock. Call this when the store is no longer needed."""
+        self._dir_lock.close()
+
     def check_directory(self, create=False):
         """
         Check if the directory exists and created it if not (and requested to do so)
@@ -69,6 +207,7 @@ def check_directory(self, create=False):
             # Create directory
             os.mkdir(self.directory_path)
             logger.info("Creating directory {}".format(self.directory_path))
+        self._acquire_dir_lock()
 
     def get_identifiable_by_hash(self, hash_: str) -> model.Identifiable:
         """
@@ -76,12 +215,13 @@ def get_identifiable_by_hash(self, hash_: str) -> model.Identifiable:
 
         :raises KeyError: If the respective file could not be found
         """
-        try:
-            with open("{}/{}.json".format(self.directory_path, hash_), "r") as file:
-                data = json.load(file, cls=json_deserialization.AASFromJsonDecoder)
-                obj = data["data"]
-        except FileNotFoundError as e:
-            raise KeyError("No Identifiable with hash {} found in local file database".format(hash_)) from e
+        with self._dir_lock.ensure_locked():
+            try:
+                with open("{}/{}.json".format(self.directory_path, hash_), "r") as file:
+                    data = json.load(file, cls=json_deserialization.AASFromJsonDecoder)
+                    obj = data["data"]
+            except FileNotFoundError as e:
+                raise KeyError("No Identifiable with hash {} found in local file database".format(hash_)) from e
         with self._object_cache_lock:
             if obj.id in self._object_cache:
                 return self._object_cache[obj.id]
@@ -94,13 +234,14 @@ def get_item(self, identifier: model.Identifier) -> model.Identifiable:
 
         :raises KeyError: If the respective file could not be found
         """
-        with self._object_cache_lock:
-            if identifier in self._object_cache:
-                return self._object_cache[identifier]
-        try:
-            return self.get_identifiable_by_hash(self._transform_id(identifier))
-        except KeyError as e:
-            raise KeyError("No Identifiable with id {} found in local file database".format(identifier)) from e
+        with self._dir_lock.ensure_locked():
+            with self._object_cache_lock:
+                if identifier in self._object_cache:
+                    return self._object_cache[identifier]
+            try:
+                return self.get_identifiable_by_hash(self._transform_id(identifier))
+            except KeyError as e:
+                raise KeyError("No Identifiable with id {} found in local file database".format(identifier)) from e
 
     def _write_atomic(self, x: model.Identifiable) -> None:
         """
@@ -128,9 +269,11 @@ def add(self, x: model.Identifiable) -> None:
         :raises KeyError: If an object with the same id exists already in the object store
         """
         logger.debug("Adding object %s to Local File Store ...", repr(x))
-        if os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))):
-            raise KeyError("Identifiable with id {} already exists in local file database".format(x.id))
-        self._write_atomic(x)
+        with self._write_lock:
+            with self._dir_lock.ensure_locked():
+                if os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))):
+                    raise KeyError("Identifiable with id {} already exists in local file database".format(x.id))
+                self._write_atomic(x)
         with self._object_cache_lock:
             self._object_cache[x.id] = x
 
@@ -141,9 +284,11 @@ def commit(self, x: model.Identifiable) -> None:
         :param x: The object to persist
         :raises KeyError: If the object is not present in the store
         """
-        if not os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))):
-            raise KeyError("No AAS object with id {} exists in local file database".format(x.id))
-        self._write_atomic(x)
+        with self._write_lock:
+            with self._dir_lock.ensure_locked():
+                if not os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))):
+                    raise KeyError("No AAS object with id {} exists in local file database".format(x.id))
+                self._write_atomic(x)
 
     def discard(self, x: model.Identifiable) -> None:
         """
@@ -153,10 +298,12 @@ def discard(self, x: model.Identifiable) -> None:
         :raises KeyError: If the object does not exist in the database
         """
         logger.debug("Deleting object %s from Local File Store database ...", repr(x))
-        try:
-            os.remove("{}/{}.json".format(self.directory_path, self._transform_id(x.id)))
-        except FileNotFoundError as e:
-            raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) from e
+        with self._write_lock:
+            with self._dir_lock.ensure_locked():
+                try:
+                    os.remove("{}/{}.json".format(self.directory_path, self._transform_id(x.id)))
+                except FileNotFoundError as e:
+                    raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) from e
         with self._object_cache_lock:
             self._object_cache.pop(x.id, None)
 
@@ -176,7 +323,8 @@ def __contains__(self, x: object) -> bool:
         else:
             return False
         logger.debug("Checking existence of object with id %s in database ...", repr(x))
-        return os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(identifier)))
+        with self._dir_lock.ensure_locked():
+            return os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(identifier)))
 
     def __len__(self) -> int:
         """
@@ -185,7 +333,8 @@ def __len__(self) -> int:
         :return: The number of objects (determined from the number of documents)
         """
         logger.debug("Fetching number of documents from database ...")
-        return sum(1 for f in os.listdir(self.directory_path) if f.lower().endswith(".json"))
+        with self._dir_lock.ensure_locked():
+            return sum(1 for f in os.listdir(self.directory_path) if f.lower().endswith(".json"))
 
     def __iter__(self) -> Iterator[model.Identifiable]:
         """
@@ -195,9 +344,10 @@ def __iter__(self) -> Iterator[model.Identifiable]:
         the identifiable objects on the fly.
         """
         logger.debug("Iterating over objects in database ...")
-        for name in os.listdir(self.directory_path):
-            if name.lower().endswith(".json"):
-                yield self.get_identifiable_by_hash(name[:-5])
+        with self._dir_lock.ensure_locked():
+            for name in os.listdir(self.directory_path):
+                if name.lower().endswith(".json"):
+                    yield self.get_identifiable_by_hash(name[:-5])
 
     @staticmethod
     def _transform_id(identifier: model.Identifier) -> str:
diff --git a/sdk/test/backend/test_local_file.py b/sdk/test/backend/test_local_file.py
index 71447f61..fee769d8 100644
--- a/sdk/test/backend/test_local_file.py
+++ b/sdk/test/backend/test_local_file.py
@@ -7,6 +7,10 @@
 import gc
 import os.path
 import shutil
+import tempfile
+import threading
+import concurrent.futures
+from typing import Callable, cast
 
 from unittest import TestCase
 
@@ -18,6 +22,106 @@
 source_core: str = "file://localhost/{}/".format(store_path)
 
 
+def run_threads(fns: list[Callable]):
+    fn_futures: list[concurrent.futures.Future] = []
+    with concurrent.futures.ThreadPoolExecutor(max_workers=len(fns)) as executor:
+        for fn in fns:
+            fn_futures.append(executor.submit(fn))
+
+    concurrent.futures.wait(fn_futures)
+    for future in fn_futures:
+        ex = future.exception()
+        if ex is not None:
+            raise ex
+
+
+class DirectoryLockTest(TestCase):
+
+    def test_double_locking(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            lock = local_file.DirectoryLock(tmpdir)
+            lock.acquire()
+            lock.acquire()
+
+            with self.assertRaises(RuntimeError):
+                lock2 = local_file.DirectoryLock(tmpdir)
+                lock2.acquire()
+
+    def test_releasing(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            lock = local_file.DirectoryLock(tmpdir)
+            lock.acquire()
+            self.assertTrue(lock._is_locked_flag)
+            lock.release()
+            self.assertFalse(lock._is_locked_flag)
+
+    def test_context_manager_fail(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            lock = local_file.DirectoryLock(tmpdir)
+
+            with self.assertRaises(RuntimeError) as cm:
+                with lock.ensure_locked():
+                    pass
+        self.assertIn("is not locked", cm.exception.args[0])
+
+    def test_context_manager_concurrency(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            lock = local_file.DirectoryLock(tmpdir)
+            lock.acquire()
+
+            barrier = threading.Barrier(parties=3, timeout=5)
+
+            def first():
+                with lock.ensure_locked():
+                    barrier.wait()
+                    barrier.wait()
+
+            def second():
+                with lock.ensure_locked():
+                    barrier.wait()
+                    barrier.wait()
+
+            def asserts():
+                barrier.wait()
+                self.assertEqual(2, lock._active_accesses)
+                barrier.wait()
+                lock.release()
+                self.assertEqual(0, lock._active_accesses)
+
+            run_threads([first, second, asserts])
+
+    def test_context_manager_finish_before_release(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            lock = local_file.DirectoryLock(tmpdir)
+            lock.acquire()
+
+            access_barrier = threading.Barrier(parties=2, timeout=5)
+            release_barrier = threading.Barrier(parties=2, timeout=5)
+
+            def access():
+                with lock.ensure_locked():
+                    access_barrier.wait()
+                    access_barrier.wait()
+                access_barrier.wait()
+
+            def release():
+                release_barrier.wait()
+                lock.release()
+                release_barrier.wait()
+
+            def asserts():
+                access_barrier.wait()
+                self.assertTrue(lock._is_locked_flag)  # in ensure_locked()
+                release_barrier.wait()
+                self.assertTrue(lock._is_locked_flag)  # in ensure_locked(), release() started
+                access_barrier.wait()
+                access_barrier.wait()
+                release_barrier.wait()
+                self.assertFalse(lock._is_locked_flag)  # out of ensure_locked(), release() finished
+
+            run_threads([access, release, asserts])
+
+
 class LocalFileBackendTest(TestCase):
     def setUp(self) -> None:
         self.identifiable_store = local_file.LocalFileIdentifiableStore(store_path)
@@ -27,8 +131,40 @@ def tearDown(self) -> None:
         try:
             self.identifiable_store.clear()
         finally:
+            self.identifiable_store.close()
             shutil.rmtree(store_path)
 
+    def test_multi_instance_fail_on_init(self):
+        # Create second store for same path and expect it to fail
+        with self.assertRaises(RuntimeError) as cm:
+            local_file.LocalFileIdentifiableStore(store_path)
+
+        self.assertIn("is already in use", cm.exception.args[0])
+
+    def test_multi_instance_fail_on_check(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = os.path.join(tmpdir, "localdb")
+
+            store1 = local_file.LocalFileIdentifiableStore(path)
+            store2 = local_file.LocalFileIdentifiableStore(path)
+
+            store1.check_directory(create=True)
+            with self.assertRaises(RuntimeError) as cm:
+                store2.check_directory()
+
+            self.assertIn("is already in use", cm.exception.args[0])
+
+    def test_dir_lock_fail_add(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = os.path.join(tmpdir, "localdb")
+            store = local_file.LocalFileIdentifiableStore(path)
+            # store did not acquire dir_lock as path does not exist yet
+
+            with self.assertRaises(RuntimeError) as cm:
+                store.add(create_example_submodel())
+
+            self.assertIn("is not locked", cm.exception.args[0])
+
     def test_identifiable_store_add(self):
         test_object = create_example_submodel()
         self.identifiable_store.add(test_object)
@@ -166,7 +302,124 @@ def test_reload_discard(self) -> None:
         example_submodel = create_example_submodel()
         self.identifiable_store.add(example_submodel)
 
-        # Reload the DictIdentifiableStore and discard the example submodel
+        # Reload the store: release the existing lock before opening a new instance
+        self.identifiable_store.close()
         self.identifiable_store = local_file.LocalFileIdentifiableStore(store_path)
         self.identifiable_store.discard(example_submodel)
         self.assertNotIn(example_submodel, self.identifiable_store)
+
+
+class _GatedLocalFileStore(local_file.LocalFileIdentifiableStore):
+    """
+    Patch :meth:`_write_atomic()` to control and enforce concurrent writes.
+    """
+
+    def __init__(self, directory_path: str, *args, **kwargs):
+        super().__init__(directory_path)
+        self._gate = threading.Event()  # gate to run _write_atomic()
+        self._gate.set()
+        self._inside = threading.Event()  # signal _write_atomic() entry
+
+    def _write_atomic(self, x: model.Identifiable) -> None:
+        self._inside.set()
+        self._gate.wait(timeout=5)
+        super()._write_atomic(x)
+
+
+class LocalFileBackendConcurrencyTest(TestCase):
+    def setUp(self):
+        self.store = _GatedLocalFileStore(store_path)
+        self.store.check_directory(create=True)
+
+    def tearDown(self):
+        try:
+            self.store.clear()
+        finally:
+            self.store.close()
+            shutil.rmtree(store_path)
+
+    def _example_submodels(self) -> tuple[model.Submodel, model.Submodel]:
+        first_submodel = model.Submodel(
+            id_='https://example.org/BackendTest',
+            submodel_element={
+                model.Property(id_short='Prop', value_type=model.datatypes.String, value='first')
+            }
+        )
+
+        second_submodel = model.Submodel(
+            id_='https://example.org/BackendTest',
+            submodel_element={
+                model.Property(id_short='Prop', value_type=model.datatypes.String, value='second')
+            }
+        )
+
+        return first_submodel, second_submodel
+
+    def test_concurrent_add(self):
+        """Checks that second add for same ID fails"""
+        first_submodel, second_submodel = self._example_submodels()
+
+        self.store._gate.clear()
+        self.store._inside.clear()
+        barrier = threading.Barrier(2, timeout=5)
+        all_done = threading.Barrier(3, timeout=5)
+
+        def first():
+            self.store.add(first_submodel)
+            all_done.wait()
+
+        def second():
+            self.store._inside.wait()
+            barrier.wait()
+            with self.assertRaises(KeyError) as ex:
+                self.store.add(second_submodel)
+
+            all_done.wait()
+            self.assertIn("already exists", ex.exception.args[0])
+
+        def control():
+            self.store._inside.wait()
+            barrier.wait()
+            self.store._gate.set()
+
+            all_done.wait()
+            submodel = self.store.get_item("https://example.org/BackendTest")
+            self.assertIsInstance(submodel, model.Submodel)
+            sm_property = submodel.get_referable("Prop")
+            self.assertIsInstance(sm_property, model.Property)
+            self.assertEqual(sm_property.value, "first")
+
+        run_threads([first, second, control])
+
+    def test_concurrent_commit_discard(self):
+        """Checks that discard is not overwritten by concurrent commit"""
+        submodel, altered_submodel = self._example_submodels()
+        self.store.add(submodel)
+
+        self.store._gate.clear()
+        self.store._inside.clear()
+        barrier = threading.Barrier(2, timeout=5)
+        all_done = threading.Barrier(3, timeout=5)
+
+        def first():
+            self.store.commit(submodel)
+            all_done.wait()
+
+        def second():
+            self.store._inside.wait()
+            barrier.wait()
+            self.store.discard(submodel)
+            all_done.wait()
+
+        def control():
+            self.store._inside.wait()
+            barrier.wait()
+            self.store._gate.set()
+
+            all_done.wait()
+            with self.assertRaises(KeyError) as ex:
+                self.store.get_item("https://example.org/BackendTest")
+
+            self.assertIn("No Identifiable", ex.exception.args[0])
+
+        run_threads([first, second, control])