diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 2d05286071..ac713b8a7c 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -4,6 +4,19 @@ Changelog
 in development
 --------------
 
+Added
+~~~~~
+
+* Added web header settings for additional security hardening to nginx.conf: X-Frame-Options,
+  Strict-Transport-Security, X-XSS-Protection and server-tokens. #5183
+
+  Contributed by @shital.
+
+* Added support for ``limit`` and ``offset`` argument to the ``list_values`` data store
+  service method (#5097 and #5171).
+
+  Contributed by @anirudhbagri.
+
 Fixed
 ~~~~~
 
@@ -24,11 +37,6 @@ Changed
 
   Contributed by @Kami and @shital.
 
-* Added web header settings for additional security hardening to nginx.conf: X-Frame-Options,
-  Strict-Transport-Security, X-XSS-Protection and server-tokens. #5183
-
-  Contributed by @shital.
-
 3.4.0 - March 02, 2021
 ----------------------
 
diff --git a/st2client/st2client/models/core.py b/st2client/st2client/models/core.py
index 83f7f59bc2..150622d08d 100644
--- a/st2client/st2client/models/core.py
+++ b/st2client/st2client/models/core.py
@@ -180,6 +180,7 @@ def get_all(self, **kwargs):
         pack = kwargs.pop("pack", None)
         prefix = kwargs.pop("prefix", None)
         user = kwargs.pop("user", None)
+        offset = kwargs.pop("offset", 0)
 
         params = kwargs.pop("params", {})
 
@@ -195,6 +196,9 @@ def get_all(self, **kwargs):
         if user:
             params["user"] = user
 
+        if offset:
+            params["offset"] = offset
+
         response = self.client.get(url=url, params=params, **kwargs)
         if response.status_code != http_client.OK:
             self.handle_error(response)
diff --git a/st2common/st2common/services/datastore.py b/st2common/st2common/services/datastore.py
index 9655499e49..d87794a9f2 100644
--- a/st2common/st2common/services/datastore.py
+++ b/st2common/st2common/services/datastore.py
@@ -68,7 +68,7 @@ def get_user_info(self):
     # Methods for datastore management
     ##################################
 
-    def list_values(self, local=True, prefix=None):
+    def list_values(self, local=True, prefix=None, limit=None, offset=0):
         """
         Retrieve all the datastores items.
 
@@ -78,13 +78,20 @@ def list_values(self, local=True, prefix=None):
         :param prefix: Optional key name prefix / startswith filter.
         :type prefix: ``str``
 
+        :param limit: Number of keys to get. Defaults to the configuration set at 'api.max_page_size'.
+        :type limit: ``integer``
+
+        :param offset: Number of keys to offset. Defaults to 0.
+        :type offset: ``integer``
+
         :rtype: ``list`` of :class:`KeyValuePair`
         """
         client = self.get_api_client()
         self._logger.debug("Retrieving all the values from the datastore")
 
+        limit = limit or cfg.CONF.api.max_page_size
         key_prefix = self._get_full_key_prefix(local=local, prefix=prefix)
-        kvps = client.keys.get_all(prefix=key_prefix)
+        kvps = client.keys.get_all(prefix=key_prefix, limit=limit, offset=offset)
         return kvps
 
     def get_value(self, name, local=True, scope=SYSTEM_SCOPE, decrypt=False):
diff --git a/st2common/tests/unit/test_datastore.py b/st2common/tests/unit/test_datastore.py
index 1e3dc86d30..9827aafd30 100644
--- a/st2common/tests/unit/test_datastore.py
+++ b/st2common/tests/unit/test_datastore.py
@@ -51,14 +51,22 @@ def test_datastore_operations_list_values(self):
         self._set_mock_api_client(mock_api_client)
 
         self._datastore_service.list_values(local=True, prefix=None)
-        mock_api_client.keys.get_all.assert_called_with(prefix="core.TestSensor:")
+        mock_api_client.keys.get_all.assert_called_with(
+            prefix="core.TestSensor:", limit=100, offset=0
+        )
         self._datastore_service.list_values(local=True, prefix="ponies")
-        mock_api_client.keys.get_all.assert_called_with(prefix="core.TestSensor:ponies")
+        mock_api_client.keys.get_all.assert_called_with(
+            prefix="core.TestSensor:ponies", limit=100, offset=0
+        )
 
         self._datastore_service.list_values(local=False, prefix=None)
-        mock_api_client.keys.get_all.assert_called_with(prefix=None)
+        mock_api_client.keys.get_all.assert_called_with(
+            prefix=None, limit=100, offset=0
+        )
         self._datastore_service.list_values(local=False, prefix="ponies")
-        mock_api_client.keys.get_all.assert_called_with(prefix="ponies")
+        mock_api_client.keys.get_all.assert_called_with(
+            prefix="ponies", limit=100, offset=0
+        )
 
         # No values in the datastore
         mock_api_client = mock.Mock()
@@ -85,6 +93,18 @@ def test_datastore_operations_list_values(self):
         self.assertEqual(len(values), 2)
         self.assertEqual(values, mock_return_value)
 
+        # Test limit
+        _ = self._datastore_service.list_values(local=True, limit=1)
+        mock_api_client.keys.get_all.assert_called_with(
+            prefix="core.TestSensor:", limit=1, offset=0
+        )
+
+        # Test offset
+        _ = self._datastore_service.list_values(local=True, offset=1)
+        mock_api_client.keys.get_all.assert_called_with(
+            prefix="core.TestSensor:", limit=100, offset=1
+        )
+
     def test_datastore_operations_get_value(self):
         mock_api_client = mock.Mock()
         kvp1 = KeyValuePair()