Repository: sopherapps/pydantic-redis (language: python)

Issue:
optional property for Model cannot be None
**Describe the bug**
If the value of an optional property is `None`, storing a `Model` fails

**To Reproduce**
Running the following code:

```python
import uuid
from typing import Optional

from pydantic import Field
from pydantic_redis import Store, RedisConfig, Model


class Child(Model):
    _primary_key_field = "id"

    id: str = Field(
        default_factory=lambda: uuid.uuid4().hex,
    )
    name: str = Field()


class Parent(Model):
    _primary_key_field = "id"
    id: str = Field(
        default_factory=lambda: uuid.uuid4().hex,
    )
    name: str = Field()
    child: Optional[Child] = Field(default=None)


store = Store(
    name="does_none_work",
    redis_config=RedisConfig(host='localhost', port=6379, db=6)
)

store.register_model(Child)
store.register_model(Parent)

parent = Parent(name="bong bing")

Parent.insert(parent)

parents_retrieved = Parent.select(ids=[parent.id])
print(parents_retrieved)
```

Fails with the following exception:
```
AttributeError: type object 'NoneType' has no attribute 'get_primary_key_field'
```

**Expected behavior**
One would expect the Parent object to be just stored into Redis with the `Optional` property left to `None`.


Discussion / hints:

Fix patch:
diff --git a/.gitignore b/.gitignore
index fb3d7f30..18f6d6c5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -145,4 +145,6 @@ cython_debug/
 
 env3.7/
 /lua_scripts/
-.DS_Store
\ No newline at end of file
+.DS_Store
+
+*.lua
\ No newline at end of file
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 36c0476c..3fdb5bec 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+### Added
+
+### Changed
+
+- Added support for dictionaries of nested models
+- Added support for deeply nested models beyond level-1 deep including:
+  - dictionaries of lists of ... of nested models
+  - lists of tuples of lists .... of nested models
+
+### Fixed
+
+- Fixed `AttributeError: 'get_primary_key_field'` when None is passed to a field with an optional nested model
+
+
 ## [0.6.0] - 2024-07-01
 
 ### Added
diff --git a/README.md b/README.md
index f64ee66e..6a459927 100644
--- a/README.md
+++ b/README.md
@@ -88,6 +88,37 @@ benchmark_bulk_insert[redis_store]                         721.2247 (6.19)     6
 ---------------------------------------------------------------------------------------------------------------------
 ```
 
+# >=v0.7 (with deeply nested models)
+
+```
+------------------------------------------------- benchmark: 22 tests -------------------------------------------------
+Name (time in us)                                              Mean                 Min                   Max          
+-----------------------------------------------------------------------------------------------------------------------
+benchmark_delete[redis_store-Wuthering Heights]            123.2946 (1.02)     107.9690 (1.0)        502.6140 (1.33)   
+benchmark_bulk_delete[redis_store]                         120.5815 (1.0)      111.9320 (1.04)       378.8660 (1.0)    
+benchmark_select_columns_for_one_id[redis_store-book2]     208.2612 (1.73)     180.4660 (1.67)       470.9860 (1.24)   
+benchmark_select_columns_for_one_id[redis_store-book1]     207.9143 (1.72)     180.6440 (1.67)       489.6890 (1.29)   
+benchmark_select_columns_for_one_id[redis_store-book0]     204.2471 (1.69)     183.4360 (1.70)       485.2500 (1.28)   
+benchmark_select_columns_for_one_id[redis_store-book3]     209.5764 (1.74)     189.5780 (1.76)       462.5650 (1.22)   
+benchmark_select_all_for_one_id[redis_store-book0]         226.4569 (1.88)     207.4920 (1.92)       499.9470 (1.32)   
+benchmark_select_all_for_one_id[redis_store-book3]         241.5488 (2.00)     210.5230 (1.95)       504.5150 (1.33)   
+benchmark_select_all_for_one_id[redis_store-book1]         234.4014 (1.94)     210.6420 (1.95)       501.2470 (1.32)   
+benchmark_select_all_for_one_id[redis_store-book2]         228.9277 (1.90)     212.0090 (1.96)       509.5740 (1.34)   
+benchmark_update[redis_store-Wuthering Heights-data0]      276.3908 (2.29)     238.3390 (2.21)       704.9450 (1.86)   
+benchmark_single_insert[redis_store-book3]                 311.0476 (2.58)     262.2940 (2.43)       589.3940 (1.56)   
+benchmark_select_columns_for_some_items[redis_store]       291.2779 (2.42)     266.0960 (2.46)       564.3510 (1.49)   
+benchmark_select_columns_paginated[redis_store]            300.4108 (2.49)     269.4740 (2.50)       552.8510 (1.46)   
+benchmark_single_insert[redis_store-book1]                 304.5771 (2.53)     274.1740 (2.54)       547.5210 (1.45)   
+benchmark_single_insert[redis_store-book2]                 317.2681 (2.63)     275.6170 (2.55)       641.5440 (1.69)   
+benchmark_single_insert[redis_store-book0]                 313.0004 (2.60)     277.3190 (2.57)       558.2160 (1.47)   
+benchmark_select_some_items[redis_store]                   343.2569 (2.85)     311.9140 (2.89)       624.6600 (1.65)   
+benchmark_select_default_paginated[redis_store]            359.8463 (2.98)     325.8310 (3.02)       623.2360 (1.65)   
+benchmark_select_columns[redis_store]                      486.6047 (4.04)     429.3250 (3.98)       867.8780 (2.29)   
+benchmark_select_default[redis_store]                      631.3835 (5.24)     584.7630 (5.42)     1,033.5990 (2.73)   
+benchmark_bulk_insert[redis_store]                         761.0832 (6.31)     724.1240 (6.71)     1,034.2950 (2.73)   
+-----------------------------------------------------------------------------------------------------------------------
+```
+
 ## Contributions
 
 Contributions are welcome. The docs have to maintained, the code has to be made cleaner, more idiomatic and faster,
diff --git a/pydantic_redis/_shared/lua_scripts.py b/pydantic_redis/_shared/lua_scripts.py
index 433c8646..8eee732b 100644
--- a/pydantic_redis/_shared/lua_scripts.py
+++ b/pydantic_redis/_shared/lua_scripts.py
@@ -1,5 +1,8 @@
 """Exposes the redis lua scripts to be used in select queries.
 
+These scripts always return a list of tuples of [record, index] where the index is a flat list of nested models
+for that record
+
 Attributes:
     SELECT_ALL_FIELDS_FOR_ALL_IDS_SCRIPT: the script for selecting all records from redis
     PAGINATED_SELECT_ALL_FIELDS_FOR_ALL_IDS_SCRIPT: the script for selecting a slice of all records from redis,
@@ -14,6 +17,8 @@
         but returning only a subset of the fields in each record.
 """
 
+# What if instead of constructing tables, we return obj as a JSON string
+
 SELECT_ALL_FIELDS_FOR_ALL_IDS_SCRIPT = """
 local s_find = string.find
 local s_gmatch = string.gmatch
@@ -29,43 +34,40 @@
     return s_find(s, prefix, 1, true) == 1
 end
 
-local function trim_dunder(s)
-    return s:match '^_*(.-)$'
-end
-
-local function get_obj(id)
+local function get_obj_and_index(id)
     local value = redis_call('HGETALL', id)
+    local idx = {}
 
     for i, k in ipairs(value) do
         if not (i % 2 == 0) then
-            if startswith(k, '___') or startswith(k, '____') then
-                local nested = {}
-
-                for v in s_gmatch(value[i + 1], '([^%[^,^%]^\"]+)') do
-                    table_insert(nested, get_obj(v))
-                end
-
-                value[i + 1] = nested
-                value[i] = trim_dunder(k)
+            if value[i + 1] == 'null' then
+            elseif startswith(k, '___') then
+               for v in s_gmatch(value[i + 1], '\"([%w_]+_%%&_[^\"%[%]]+)\"') do
+                   table_insert(idx, v)
+                   table_insert(idx, {get_obj_and_index(v)})
+               end
             elseif startswith(k, '__') then
-                value[i + 1] = get_obj(value[i + 1])
-                value[i] = trim_dunder(k)
+               table_insert(idx, value[i + 1])
+               table_insert(idx, {get_obj_and_index(value[i + 1])})
             end
-        end
+            
+       end
     end
 
     if next(value) == nil then
-        return id
+        return id, nil
     end
-
-    return value
+    return value, idx
 end
 
 repeat
     local result = redis_call('SCAN', cursor, 'MATCH', ARGV[1])
     for _, key in ipairs(result[2]) do
         if redis_call('TYPE', key).ok == 'hash' then
-            table_insert(filtered, get_obj(key))
+            local value, idx = get_obj_and_index(key)
+            if type(value) == 'table' then
+                table_insert(filtered, {value, idx})
+            end
         end
     end
     cursor = result[1]
@@ -85,36 +87,29 @@
     return s_find(s, prefix, 1, true) == 1
 end
 
-local function trim_dunder(s)
-    return s:match '^_*(.-)$'
-end
-
-local function get_obj(id)
+local function get_obj_and_index(id)
     local value = redis_call('HGETALL', id)
+    local idx = {}
 
     for i, k in ipairs(value) do
         if not (i % 2 == 0) then
-            if startswith(k, '___') or startswith(k, '____') then
-                local nested = {}
-
-                for v in s_gmatch(value[i + 1], '([^%[^,^%]^\"]+)') do
-                    table_insert(nested, get_obj(v))
-                end
-
-                value[i + 1] = nested
-                value[i] = trim_dunder(k)
+            if value[i + 1] == 'null' then
+            elseif startswith(k, '___') then
+               for v in s_gmatch(value[i + 1], '\"([%w_]+_%%&_[^\"%[%]]+)\"') do
+                   table_insert(idx, v)
+                   table_insert(idx, {get_obj_and_index(v)})
+               end
             elseif startswith(k, '__') then
-                value[i + 1] = get_obj(value[i + 1])
-                value[i] = trim_dunder(k)
+               table_insert(idx, value[i + 1])
+               table_insert(idx, {get_obj_and_index(value[i + 1])})
             end
-        end
+       end
     end
 
     if next(value) == nil then
-        return id
+        return id, nil
     end
-
-    return value
+    return value, idx
 end
 
 local table_index_key = ARGV[1]
@@ -125,9 +120,9 @@
 local ids = redis_call('ZRANGE', table_index_key, start, stop)
 
 for _, key in ipairs(ids) do
-    local value = get_obj(key)
+    local value, idx = get_obj_and_index(key)
     if type(value) == 'table' then
-        table_insert(result, value)
+        table_insert(result, {value, idx})
     end
 end
 
@@ -148,42 +143,35 @@
     return s_find(s, prefix, 1, true) == 1
 end
 
-local function trim_dunder(s)
-    return s:match '^_*(.-)$'
-end
-
-local function get_obj(id)
+local function get_obj_and_index(id)
     local value = redis_call('HGETALL', id)
+    local idx = {}
 
     for i, k in ipairs(value) do
         if not (i % 2 == 0) then
-            if startswith(k, '___') or startswith(k, '____') then
-                local nested = {}
-
-                for v in s_gmatch(value[i + 1], '([^%[^,^%]^\"]+)') do
-                    table_insert(nested, get_obj(v))
-                end
-
-                value[i + 1] = nested
-                value[i] = trim_dunder(k)
+            if value[i + 1] == 'null' then
+            elseif startswith(k, '___') then
+               for v in s_gmatch(value[i + 1], '\"([%w_]+_%%&_[^\"%[%]]+)\"') do
+                   table_insert(idx, v)
+                   table_insert(idx, {get_obj_and_index(v)})
+               end
             elseif startswith(k, '__') then
-                value[i + 1] = get_obj(value[i + 1])
-                value[i] = trim_dunder(k)
+               table_insert(idx, value[i + 1])
+               table_insert(idx, {get_obj_and_index(value[i + 1])})
             end
-        end
+       end
     end
 
     if next(value) == nil then
-        return id
+        return id, nil
     end
-
-    return value
+    return value, idx
 end
 
 for _, key in ipairs(KEYS) do
-    local value = get_obj(key)
+    local value, idx = get_obj_and_index(key)
     if type(value) == 'table' then
-        table_insert(result, value)
+        table_insert(result, {value, idx})
     end
 end
 
@@ -207,36 +195,29 @@
     return s_find(s, prefix, 1, true) == 1
 end
 
-local function trim_dunder(s)
-    return s:match '^_*(.-)$'
-end
-
-local function get_obj(id)
+local function get_obj_and_index(id)
     local value = redis_call('HGETALL', id)
+    local idx = {}
 
     for i, k in ipairs(value) do
         if not (i % 2 == 0) then
-            if startswith(k, '___') or startswith(k, '____') then
-                local nested = {}
-
-                for v in s_gmatch(value[i + 1], '([^%[^,^%]^\"]+)') do
-                    table_insert(nested, get_obj(v))
-                end
-
-                value[i + 1] = nested
-                value[i] = trim_dunder(k)
+            if value[i + 1] == 'null' then
+            elseif startswith(k, '___') then
+               for v in s_gmatch(value[i + 1], '\"([%w_]+_%%&_[^\"%[%]]+)\"') do
+                   table_insert(idx, v)
+                   table_insert(idx, {get_obj_and_index(v)})
+               end
             elseif startswith(k, '__') then
-                value[i + 1] = get_obj(value[i + 1])
-                value[i] = trim_dunder(k)
+               table_insert(idx, value[i + 1])
+               table_insert(idx, {get_obj_and_index(value[i + 1])})
             end
-        end
+       end
     end
 
     if next(value) == nil then
-        return id
+        return id, nil
     end
-
-    return value
+    return value, idx
 end
 
 for i, k in ipairs(ARGV) do
@@ -251,13 +232,20 @@
         if redis_call('TYPE', key).ok == 'hash' then
             local data = redis_call('HMGET', key, table_unpack(columns))
             local parsed_data = {}
+            local index = {}
 
             for i, v in ipairs(data) do
-                table_insert(parsed_data, trim_dunder(columns[i]))
-                table_insert(parsed_data, get_obj(v))
+                table_insert(parsed_data, columns[i])
+                table_insert(parsed_data, v)
+                    
+                local value, idx = get_obj_and_index(v)
+                if type(idx) == 'table' then
+                    table_insert(index, v)
+                    table_insert(index, {value, idx})
+                end
             end
-
-            table_insert(filtered, parsed_data)
+        
+            table_insert(filtered, {parsed_data, index})
         end
     end
     cursor = result[1]
@@ -278,36 +266,29 @@
     return s_find(s, prefix, 1, true) == 1
 end
 
-local function trim_dunder(s)
-    return s:match '^_*(.-)$'
-end
-
-local function get_obj(id)
+local function get_obj_and_index(id)
     local value = redis_call('HGETALL', id)
+    local idx = {}
 
     for i, k in ipairs(value) do
         if not (i % 2 == 0) then
-            if startswith(k, '___') or startswith(k, '____') then
-                local nested = {}
-
-                for v in s_gmatch(value[i + 1], '([^%[^,^%]^\"]+)') do
-                    table_insert(nested, get_obj(v))
-                end
-
-                value[i + 1] = nested
-                value[i] = trim_dunder(k)
+            if value[i + 1] == 'null' then
+            elseif startswith(k, '___') then
+               for v in s_gmatch(value[i + 1], '\"([%w_]+_%%&_[^\"%[%]]+)\"') do
+                   table_insert(idx, v)
+                   table_insert(idx, {get_obj_and_index(v)})
+               end
             elseif startswith(k, '__') then
-                value[i + 1] = get_obj(value[i + 1])
-                value[i] = trim_dunder(k)
+               table_insert(idx, value[i + 1])
+               table_insert(idx, {get_obj_and_index(value[i + 1])})
             end
-        end
+       end
     end
 
     if next(value) == nil then
-        return id
+        return id, nil
     end
-
-    return value
+    return value, idx
 end
 
 local result = {}
@@ -327,15 +308,23 @@
 for _, key in ipairs(ids) do
     local data = redis_call('HMGET', key, table_unpack(columns))
     local parsed_data = {}
+    local index = {}
 
     for i, v in ipairs(data) do
         if v then
-            table_insert(parsed_data, trim_dunder(columns[i]))
-            table_insert(parsed_data, get_obj(v))
+            table_insert(parsed_data, columns[i])
+            table_insert(parsed_data, v)
+            
+            local value, idx = get_obj_and_index(v)
+            if type(idx) == 'table' then
+                table_insert(index, v)
+                table_insert(index, {value, idx})
+            end
         end
     end
 
-    table_insert(result, parsed_data)
+    table_insert(result, {parsed_data, index})
+
 end
 
 return result
@@ -357,36 +346,29 @@
     return s_find(s, prefix, 1, true) == 1
 end
 
-local function trim_dunder(s)
-    return s:match '^_*(.-)$'
-end
-
-local function get_obj(id)
+local function get_obj_and_index(id)
     local value = redis_call('HGETALL', id)
+    local idx = {}
 
     for i, k in ipairs(value) do
         if not (i % 2 == 0) then
-            if startswith(k, '___') or startswith(k, '____') then
-                local nested = {}
-
-                for v in s_gmatch(value[i + 1], '([^%[^,^%]^\"]+)') do
-                    table_insert(nested, get_obj(v))
-                end
-
-                value[i + 1] = nested
-                value[i] = trim_dunder(k)
+            if value[i + 1] == 'null' then
+            elseif startswith(k, '___') then
+               for v in s_gmatch(value[i + 1], '\"([%w_]+_%%&_[^\"%[%]]+)\"') do
+                   table_insert(idx, v)
+                   table_insert(idx, {get_obj_and_index(v)})
+               end
             elseif startswith(k, '__') then
-                value[i + 1] = get_obj(value[i + 1])
-                value[i] = trim_dunder(k)
+               table_insert(idx, value[i + 1])
+               table_insert(idx, {get_obj_and_index(value[i + 1])})
             end
-        end
+       end
     end
 
     if next(value) == nil then
-        return id
+        return id, nil
     end
-
-    return value
+    return value, idx
 end
 
 for _, k in ipairs(ARGV) do
@@ -396,15 +378,22 @@
 for _, key in ipairs(KEYS) do
     local data = redis_call('HMGET', key, table_unpack(columns))
     local parsed_data = {}
+    local index = {}
 
     for i, v in ipairs(data) do
         if v then
-            table_insert(parsed_data, trim_dunder(columns[i]))
-            table_insert(parsed_data, get_obj(v))
+            table_insert(parsed_data, columns[i])
+            table_insert(parsed_data, v)
+            
+            local value, idx = get_obj_and_index(v)
+            if type(idx) == 'table' then
+                table_insert(index, v)
+                table_insert(index, {value, idx})
+            end
         end
     end
 
-    table_insert(result, parsed_data)
+    table_insert(result, {parsed_data, index})
 end
 return result
 """
diff --git a/pydantic_redis/_shared/model/base.py b/pydantic_redis/_shared/model/base.py
index 2d52d501..00c1f545 100644
--- a/pydantic_redis/_shared/model/base.py
+++ b/pydantic_redis/_shared/model/base.py
@@ -1,12 +1,17 @@
 """Exposes the Base `Model` common to both async and sync APIs
 
+Attributes:
+    NESTED_MODEL_PREFIX (str): the prefix for fields with single nested models
+    NESTED_MODEL_LIST_FIELD_PREFIX (str): the prefix for fields with lists of nested models
+    NESTED_MODEL_TUPLE_FIELD_PREFIX (str): the prefix for fields with tuples of nested models
+    NESTED_MODEL_DICT_FIELD_PREFIX (str): the prefix for fields with dicts of nested models
 """
 
+import enum
 import typing
 from typing import Dict, Tuple, Any, Type, Union, List, Optional
 
 from pydantic import ConfigDict, BaseModel
-from pydantic.fields import ModelPrivateAttr
 
 from pydantic_redis._shared.utils import (
     typing_get_origin,
@@ -15,11 +20,40 @@
     from_dict_to_key_value_list,
     from_bytes_to_str,
     from_str_or_bytes_to_any,
+    groups_of_n,
 )
 
 
 from ..store import AbstractStore
 
+NESTED_MODEL_PREFIX = "__"
+NESTED_MODEL_LIST_FIELD_PREFIX = "___"
+NESTED_MODEL_TUPLE_FIELD_PREFIX = "____"
+NESTED_MODEL_DICT_FIELD_PREFIX = "_____"
+
+
+class NestingType(int, enum.Enum):
+    """The type of nesting that can happen especially for nested models"""
+
+    ON_ROOT = 0
+    IN_LIST = 1
+    IN_TUPLE = 2
+    IN_DICT = 3
+    IN_UNION = 4
+
+
+# the type describing a tree for traversing types that form an aggregate type with a possibility
+#   of nested types and models
+#   Note: AbstractModel types are treated special. The first item in the tuple declares if
+#         the type on that tree node has a nested model, and of which type of nesting
+#
+#   Note: (None, (Any)) corresponds to a type that is not a nested model
+#         (IN_LIST, (AbstractModel)) corresponds to List[AbstractModel]
+#         (None, (str)) corresponds to str
+#         (IN_TUPLE, (str, AbstractModel)) corresponds to Tuple[str, AbstractModel]
+#         (IN_LIST, (IN_TUPLE, (str, AbstractModel))) corresponds to List[Tuple[str, AbstractModel]]
+AggTypeTree = Tuple[Optional[NestingType], Tuple[Union[Type["AbstractModel"], Any], ...]]  # type: ignore
+
 
 class AbstractModel(BaseModel):
     """A base class for all Models, sync and async alike.
@@ -32,21 +66,18 @@ class AbstractModel(BaseModel):
         _field_types (Dict[str, Any]): a mapping of the fields and their types for
             the current model
         _store (AbstractStore): the Store in which the current model is registered.
-        _nested_model_tuple_fields (Dict[str, Tuple[Any, ...]]): a mapping of
-            fields and their types for fields that have tuples of nested models
-        _nested_model_list_fields (Dict[str, Type["AbstractModel"]]): a mapping of
-            fields and their associated nested models for fields that have
-            lists of nested models
-        _nested_model_fields (Dict[str, Type["AbstractModel"]]): a mapping of
-            fields and their associated nested models for fields that have nested models
+        _field_type_trees (Dict[str, Optional[AggTypeTree]]): a mapping of
+            fields and their associated trees of types forming their aggregate types
+        _strict (bool): Whether the model should be very strict on its types. By default, a
+            moderate level of strictness is imposed
     """
 
     _primary_key_field: str
     _field_types: Dict[str, Any] = {}
     _store: AbstractStore
-    _nested_model_tuple_fields: Dict[str, Tuple[Any, ...]] = {}
-    _nested_model_list_fields: Dict[str, Type["AbstractModel"]] = {}
-    _nested_model_fields: Dict[str, Type["AbstractModel"]] = {}
+    _field_type_trees: Dict[str, Optional[AggTypeTree]] = {}
+    _field_typed_keys: Dict[str, str] = {}
+    _strict: bool = False
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     @classmethod
@@ -59,32 +90,13 @@ def get_store(cls) -> AbstractStore:
         return cls._store
 
     @classmethod
-    def get_nested_model_tuple_fields(cls):
-        """Gets the mapping for fields that have tuples of nested models.
-
-        Returns:
-             The mapping of field name and field type of a form similar to
-              `Tuple[str, Book, date]`
-        """
-        return cls._nested_model_tuple_fields
-
-    @classmethod
-    def get_nested_model_list_fields(cls):
-        """Gets the mapping for fields that have lists of nested models.
+    def get_field_type_trees(cls):
+        """Gets the mapping for fields and the trees of the types that form their aggregate types.
 
         Returns:
-             The mapping of field name and model class nested in that field.
+             The mapping of field name and type trees of their aggregate types
         """
-        return cls._nested_model_list_fields
-
-    @classmethod
-    def get_nested_model_fields(cls):
-        """Gets the mapping for fields that have nested models.
-
-        Returns:
-             The mapping of field name and model class nested in that field.
-        """
-        return cls._nested_model_fields
+        return cls._field_type_trees
 
     @classmethod
     def get_primary_key_field(cls):
@@ -107,6 +119,15 @@ def get_field_types(cls) -> Dict[str, Any]:
         """
         return cls._field_types
 
+    @classmethod
+    def get_field_typed_keys(cls) -> Dict[str, Any]:
+        """Gets the mapping of field and their type-aware key names for current Model.
+
+        Returns:
+            the mapping of field and the type-aware key names for current Model
+        """
+        return cls._field_typed_keys
+
     @classmethod
     def initialize(cls):
         """Initializes class-wide variables for performance's reasons.
@@ -116,48 +137,16 @@ def initialize(cls):
         """
         cls._field_types = typing.get_type_hints(cls)
 
-        cls._nested_model_list_fields = {}
-        cls._nested_model_tuple_fields = {}
-        cls._nested_model_fields = {}
+        cls._field_type_trees = {
+            field: _generate_field_type_tree(field_type, strict=cls._strict)
+            for field, field_type in cls._field_types.items()
+            if not field.startswith("_")
+        }
 
-        for field, field_type in cls._field_types.items():
-            try:
-                # In case the annotation is Optional, an alias of Union[X, None], extract the X
-                is_generic = hasattr(field_type, "__origin__")
-                if (
-                    is_generic
-                    and typing_get_origin(field_type) == Union
-                    and typing_get_args(field_type)[-1] == None.__class__
-                ):
-                    field_type = typing_get_args(field_type)[0]
-                    is_generic = hasattr(field_type, "__origin__")
-
-                if (
-                    is_generic
-                    and typing_get_origin(field_type) in (List, list)
-                    and issubclass(typing_get_args(field_type)[0], AbstractModel)
-                ):
-                    cls._nested_model_list_fields[field] = typing_get_args(field_type)[
-                        0
-                    ]
-
-                elif (
-                    is_generic
-                    and typing_get_origin(field_type) in (Tuple, tuple)
-                    and any(
-                        [
-                            issubclass(v, AbstractModel)
-                            for v in typing_get_args(field_type)
-                        ]
-                    )
-                ):
-                    cls._nested_model_tuple_fields[field] = typing_get_args(field_type)
-
-                elif issubclass(field_type, AbstractModel):
-                    cls._nested_model_fields[field] = field_type
-
-            except (TypeError, AttributeError):
-                pass
+        cls._field_typed_keys = {
+            field: _get_typed_field_key(field, type_tree=type_tree)
+            for field, type_tree in cls._field_type_trees.items()
+        }
 
     @classmethod
     def serialize_partially(cls, data: Optional[Dict[str, Any]]) -> Dict[str, Any]:
@@ -173,89 +162,210 @@ def serialize_partially(cls, data: Optional[Dict[str, Any]]) -> Dict[str, Any]:
 
     @classmethod
     def deserialize_partially(
-        cls, data: Union[List[Any], Dict[Any, Any]] = ()
+        cls, data: Union[List[Any], Dict[Any, Any]] = (), index: Dict[Any, Any] = None
     ) -> Dict[str, Any]:
-        """Casts str or bytes in a dict or flattened key-value list to expected data types.
+        """Casts str or bytes in a dict to expected data types.
 
         Converts str or bytes to their expected data types
 
         Args:
             data: flattened list of key-values or dictionary of data to cast.
                 Keeping it as potentially a dictionary ensures backward compatibility.
+            index: dictionary of the index of nested models potentially present
 
         Returns:
             the dictionary of properly parsed key-values.
         """
+        if index is None:
+            index = {}
+
         if isinstance(data, dict):
             # for backward compatibility
             data = from_dict_to_key_value_list(data)
 
         parsed_dict = {}
 
-        nested_model_list_fields = cls.get_nested_model_list_fields()
-        nested_model_tuple_fields = cls.get_nested_model_tuple_fields()
-        nested_model_fields = cls.get_nested_model_fields()
+        field_type_trees = cls.get_field_type_trees()
 
-        for i in range(0, len(data), 2):
-            key = from_bytes_to_str(data[i])
+        for k, v in groups_of_n(data, 2):
+            # remove the dunders for nested model fields
+            key = from_bytes_to_str(k).lstrip("_")
             field_type = cls._field_types.get(key)
-            value = from_str_or_bytes_to_any(value=data[i + 1], field_type=field_type)
-
-            if key in nested_model_list_fields and value is not None:
-                value = _cast_lists(value, nested_model_list_fields[key])
-
-            elif key in nested_model_tuple_fields and value is not None:
-                value = _cast_tuples(value, nested_model_tuple_fields[key])
+            value = from_str_or_bytes_to_any(value=v, field_type=field_type)
+            type_tree = field_type_trees.get(key)
 
-            elif key in nested_model_fields and value is not None:
-                value = _cast_to_model(value=value, model=nested_model_fields[key])
-
-            parsed_dict[key] = value
+            parsed_dict[key] = _cast_by_type_tree(
+                value=value, type_tree=type_tree, index=index
+            )
 
         return parsed_dict
 
 
-def _cast_lists(value: List[Any], _type: Type[AbstractModel]) -> List[AbstractModel]:
-    """Casts a list of flattened key-value lists into a list of _type.
+def _generate_field_type_tree(field_type: Any, strict: bool = False) -> AggTypeTree:
+    """Gets the tree of types for the given aggregate type of the field
 
     Args:
-        _type: the type to cast the records to.
-        value: the value to convert
+        field_type: the type of the field
+        strict: whether to raise an error if a given generic type is not supported; default=False
 
     Returns:
-        a list of records of the given _type
+        the type of nested model or None if not a nested model instance,
+            and a tuple of the types of its constituent types
     """
-    return [_type(**_type.deserialize_partially(item)) for item in value]
+    try:
+        nesting_type = None
+        generic_cls = typing_get_origin(field_type)
+
+        if generic_cls is None:
+            if issubclass(field_type, AbstractModel):
+                return NestingType.ON_ROOT, (field_type,)
+            return None, (field_type,)
+
+        type_args = typing_get_args(field_type)
+
+        if generic_cls is Union:
+            nesting_type = NestingType.IN_UNION
+
+        elif generic_cls in (List, list):
+            nesting_type = NestingType.IN_LIST
 
+        elif generic_cls in (Tuple, tuple):
+            nesting_type = NestingType.IN_TUPLE
 
-def _cast_tuples(value: List[Any], _type: Tuple[Any, ...]) -> Tuple[Any, ...]:
-    """Casts a list of flattened key-value lists into a list of tuple of _type,.
+        elif generic_cls in (Dict, dict):
+            nesting_type = NestingType.IN_DICT
+
+        elif strict:
+            raise NotImplementedError(
+                f"Generic class type: {generic_cls} not supported for nested models"
+            )
+
+        return nesting_type, tuple(
+            [_generate_field_type_tree(v, strict) for v in type_args]
+        )
+
+    except AttributeError:
+        return None, (field_type,)
+
+
+def _cast_by_type_tree(
+    value: Any, type_tree: Optional[AggTypeTree], index: Dict[Any, Any] = None
+) -> Any:
+    """Casts a given value into a value basing on the tree of its aggregate type
 
     Args:
-        _type: the tuple signature type to cast the records to
-            e.g. Tuple[str, Book, int]
-        value: the value to convert
+        value: the value to be cast basing on the type tree
+        type_tree: the tree representing the nested hierarchy of types for the aggregate
+            type that the value is to be cast into
+        index: dictionary of the index of nested models potentially present
 
     Returns:
-        a list of records of tuple signature specified by `_type`
+        the parsed value
     """
-    items = []
-    for field_type, value in zip(_type, value):
-        if issubclass(field_type, AbstractModel) and value is not None:
-            value = field_type(**field_type.deserialize_partially(value))
-        items.append(value)
+    if value is None or type_tree is None:
+        # return the value as is because it cannot be cast
+        return value
+
+    nesting_type, type_args = type_tree
+
+    if nesting_type is NestingType.ON_ROOT:
+        _type = type_args[0]
+        nested_model_data = value
+        if isinstance(value, str):
+            # load the nested model if it is not yet loaded
+            nested_model_data = index.get(value, value)
+        return _type(**_type.deserialize_partially(nested_model_data))
+
+    if nesting_type is NestingType.IN_LIST:
+        _type = type_args[0]
+        return [_cast_by_type_tree(item, _type, index) for item in value]
+
+    if nesting_type is NestingType.IN_TUPLE:
+        return tuple(
+            [
+                _cast_by_type_tree(item, _type, index)
+                for _type, item in zip(type_args, value)
+            ]
+        )
+
+    if nesting_type is NestingType.IN_DICT:
+        _, value_type = type_args
+        return {k: _cast_by_type_tree(v, value_type, index) for k, v in value.items()}
+
+    if nesting_type is NestingType.IN_UNION:
+        # the value can be any of the types in type_args
+        for _type in type_args:
+            try:
+                parsed_value = _cast_by_type_tree(value, _type, index)
+                # return the first successfully parsed value
+                # that is not equal to the original value
+                if parsed_value != value:
+                    return parsed_value
+            except Exception:
+                pass
 
-    return tuple(items)
+    # return the value without any parsing
+    return value
 
 
-def _cast_to_model(value: List[Any], model: Type[AbstractModel]) -> AbstractModel:
-    """Converts a list of flattened key-value lists into a list of models,.
+def _get_typed_field_key(
+    field: str, type_tree: AggTypeTree, initial_prefix: str = ""
+) -> str:
+    """Returns the key for the given field with extra information of its type
 
     Args:
-        model: the model class to cast to
-        value: the value to cast
+        field: the original key
+        type_tree: the tree of types that form the aggregate tree for the given key
+        initial_prefix: the initial_prefix to add to the field
 
     Returns:
-        a list of model instances of type `model`
+        the key with extra type information in its string
     """
-    return model(**model.deserialize_partially(value))
+    if type_tree is None:
+        return field
+
+    nesting_type, type_args = type_tree
+
+    if nesting_type is NestingType.ON_ROOT:
+        # we have found a NestedModel, so we stop recursion
+        if initial_prefix:
+            return f"{initial_prefix}{field}"
+        return f"{NESTED_MODEL_PREFIX}{field}"
+
+    if nesting_type is NestingType.IN_LIST:
+        _type = type_args[0]
+        return _get_typed_field_key(
+            field, _type, initial_prefix=NESTED_MODEL_LIST_FIELD_PREFIX
+        )
+
+    if nesting_type is NestingType.IN_TUPLE:
+        for _type in type_args:
+            key = _get_typed_field_key(
+                field, _type, initial_prefix=NESTED_MODEL_TUPLE_FIELD_PREFIX
+            )
+            if key != field:
+                return key
+
+    if nesting_type is NestingType.IN_DICT:
+        _, value_type = type_args
+        return _get_typed_field_key(
+            field, value_type, initial_prefix=NESTED_MODEL_DICT_FIELD_PREFIX
+        )
+
+    if nesting_type is NestingType.IN_UNION:
+        # the value can be any of the types in type_args
+        for _type in type_args:
+            try:
+                key = _get_typed_field_key(field, type_tree=_type)
+                # return the first successful value
+                # that is not equal to the original value
+                # FIXME: Note that this is not comprehensive enough because
+                #   it is possible to have Union[AbstractModel, List[AbstractModel]]
+                #   but that would be a complicated type to parse/serialize
+                #   Moral of story: Don't use it :-)
+                if key != field:
+                    return key
+            except Exception:
+                pass
+
+    return field
diff --git a/pydantic_redis/_shared/model/insert_utils.py b/pydantic_redis/_shared/model/insert_utils.py
index aae524bc..6d87122e 100644
--- a/pydantic_redis/_shared/model/insert_utils.py
+++ b/pydantic_redis/_shared/model/insert_utils.py
@@ -3,7 +3,7 @@
 """
 
 from datetime import datetime
-from typing import Union, Optional, Any, Dict, Tuple, List, Type
+from typing import Union, Optional, Any, Dict, Type
 
 from redis.asyncio.client import Pipeline as AioPipeline
 from redis.client import Pipeline
@@ -11,12 +11,14 @@
 from .prop_utils import (
     get_redis_key,
     get_model_index_key,
-    NESTED_MODEL_PREFIX,
-    NESTED_MODEL_LIST_FIELD_PREFIX,
-    NESTED_MODEL_TUPLE_FIELD_PREFIX,
 )
 
-from .base import AbstractModel
+from .base import (
+    AbstractModel,
+    NestingType,
+    AggTypeTree,
+    NESTED_MODEL_LIST_FIELD_PREFIX,
+)
 
 
 def insert_on_pipeline(
@@ -75,16 +77,16 @@ def _serialize_nested_models(
 
     In order to make the record serializable, all nested models including those in
     lists and tuples of nested models are converted to their primary keys,
-    after being their insert operations have been added to the pipeline.
+    after their insert operations have been added to the pipeline.
 
     A few cleanups it does include:
       - Upserting any nested records in `record`
       - Replacing the keys of nested records with their NESTED_MODEL_PREFIX prefixed versions
         e.g. `__author` instead of author
       - Replacing the keys of lists of nested records with their NESTED_MODEL_LIST_FIELD_PREFIX prefixed versions
-        e.g. `__%&l_author` instead of author
+        e.g. `___author` instead of author
       - Replacing the keys of tuples of nested records with their NESTED_MODEL_TUPLE_FIELD_PREFIX prefixed versions
-        e.g. `__%&l_author` instead of author
+        e.g. `____author` instead of author
       - Replacing the values of nested records with their foreign keys
 
     Args:
@@ -97,136 +99,97 @@ def _serialize_nested_models(
         the partially serialized dict that has no nested models
     """
     data = record.items() if isinstance(record, dict) else record
-    new_data = {}
-
-    nested_model_list_fields = model.get_nested_model_list_fields()
-    nested_model_tuple_fields = model.get_nested_model_tuple_fields()
-    nested_model_fields = model.get_nested_model_fields()
+    field_type_trees = model.get_field_type_trees()
+    field_typed_keys = model.get_field_typed_keys()
 
+    new_data = {}
     for k, v in data:
-        key, value = k, v
-
-        if key in nested_model_list_fields:
-            key, value = _serialize_list(
-                key=key, value=value, pipeline=pipeline, life_span=life_span
-            )
-        elif key in nested_model_tuple_fields:
-            key, value = _serialize_tuple(
-                key=key,
-                value=value,
-                pipeline=pipeline,
-                life_span=life_span,
-                tuple_fields=nested_model_tuple_fields,
-            )
-        elif key in nested_model_fields:
-            key, value = _serialize_model(
-                key=key, value=value, pipeline=pipeline, life_span=life_span
-            )
+        type_tree = field_type_trees.get(k)
+        key = field_typed_keys.get(k, k)
+        new_data[key] = _serialize_by_type_tree(
+            value=v, type_tree=type_tree, pipeline=pipeline, life_span=life_span
+        )
 
-        new_data[key] = value
     return new_data
 
 
-def _serialize_tuple(
-    key: str,
-    value: Tuple[AbstractModel],
+def _serialize_by_type_tree(
+    value: Any,
+    type_tree: Optional[AggTypeTree],
     pipeline: Union[Pipeline, AioPipeline],
     life_span: Optional[Union[float, int]],
-    tuple_fields: Dict[str, Tuple[Any, ...]],
-) -> Tuple[str, List[Any]]:
-    """Replaces models in a tuple with strings.
-
-    It adds insert operations for the records in the tuple onto the pipeline
-    and returns the tuple with the models replaced by their primary keys as value.
-
-    Returns:
-        key: the original `key` prefixed with NESTED_MODEL_TUPLE_FIELD_PREFIX
-        value: tthe tuple with the models replaced by their primary keys
-    """
-    try:
-        field_types = tuple_fields.get(key, ())
-        value = [
-            (
-                insert_on_pipeline(
-                    model=field_type,
-                    _id=None,
-                    pipeline=pipeline,
-                    record=item,
-                    life_span=life_span,
-                )
-                if issubclass(field_type, AbstractModel)
-                else item
-            )
-            for field_type, item in zip(field_types, value)
-        ]
-        key = f"{NESTED_MODEL_TUPLE_FIELD_PREFIX}{key}"
-    except TypeError:
-        # In case the value is None, just ignore
-        pass
-
-    return key, value
-
+) -> Any:
+    """Transforms a given value into a value basing on the tree of its aggregate type
 
-def _serialize_list(
-    key: str,
-    value: List[AbstractModel],
-    pipeline: Union[Pipeline, AioPipeline],
-    life_span: Optional[Union[float, int]],
-) -> Tuple[str, List[Any]]:
-    """Casts a list of models into a list of strings
+    Nested models are inserted into the redis database and their positions in the data
+    replaced by their primary keys
 
-    It adds insert operations for the records in the list onto the pipeline
-    and returns a list of their primary keys as value.
+    Args:
+        value: the value to be serialized basing on the type tree
+        type_tree: the tree representing the nested hierarchy of types for the aggregate
+            type that the value is to be cast into
+        pipeline: the redis pipeline on which the redis operations are to be done.
+        life_span: the time-to-live in seconds for the given record.
 
     Returns:
-        key: the original `key` prefixed with NESTED_MODEL_LIST_FIELD_PREFIX
-        value: the list of primary keys of the records to be inserted
+        the serialized value
     """
-    try:
-        value = [
-            insert_on_pipeline(
-                model=item.__class__,
-                _id=None,
-                pipeline=pipeline,
-                record=item,
-                life_span=life_span,
-            )
-            for item in value
-        ]
-        key = f"{NESTED_MODEL_LIST_FIELD_PREFIX}{key}"
-    except TypeError:
-        # In case the value is None, just ignore
-        pass
-
-    return key, value
-
-
-def _serialize_model(
-    key: str,
-    value: AbstractModel,
-    pipeline: Union[Pipeline, AioPipeline],
-    life_span: Optional[Union[float, int]],
-) -> Tuple[str, str]:
-    """Casts a model into a string
+    if type_tree is None:
+        # return the value as is because it cannot be serialized
+        return value
 
-    It adds an insert operation for the given model onto the pipeline
-    and returns its primary key as value.
+    nesting_type, type_args = type_tree
 
-    Returns:
-        key: the original `key` prefixed with NESTED_MODEL_PREFIX
-        value: the primary key of the model
-    """
-    try:
-        value = insert_on_pipeline(
+    if nesting_type is NestingType.ON_ROOT:
+        return insert_on_pipeline(
             model=value.__class__,
             _id=None,
             pipeline=pipeline,
             record=value,
             life_span=life_span,
         )
-        key = f"{NESTED_MODEL_PREFIX}{key}"
-    except TypeError:
-        # In case the value is None, just ignore
-        pass
 
-    return key, value
+    if nesting_type is NestingType.IN_LIST:
+        _type = type_args[0]
+        return [
+            _serialize_by_type_tree(
+                value=item, type_tree=_type, pipeline=pipeline, life_span=life_span
+            )
+            for item in value
+        ]
+
+    if nesting_type is NestingType.IN_TUPLE:
+        return tuple(
+            [
+                _serialize_by_type_tree(
+                    value=item, type_tree=_type, pipeline=pipeline, life_span=life_span
+                )
+                for _type, item in zip(type_args, value)
+            ]
+        )
+
+    if nesting_type is NestingType.IN_DICT:
+        _, value_type = type_args
+        return {
+            k: _serialize_by_type_tree(
+                value=v, type_tree=value_type, pipeline=pipeline, life_span=life_span
+            )
+            for k, v in value.items()
+        }
+
+    if nesting_type is NestingType.IN_UNION:
+        # the value can be any of the types in type_args
+        for _type in type_args:
+            try:
+                serialized_value = _serialize_by_type_tree(
+                    value=value, type_tree=_type, pipeline=pipeline, life_span=life_span
+                )
+                # return the first successfully serialized value
+                # that is not equal to the original value
+                if serialized_value != value:
+                    return serialized_value
+            except Exception:
+                pass
+
+    # return the value without any serializing
+    return value
diff --git a/pydantic_redis/_shared/model/prop_utils.py b/pydantic_redis/_shared/model/prop_utils.py
index 9256a247..8eb01544 100644
--- a/pydantic_redis/_shared/model/prop_utils.py
+++ b/pydantic_redis/_shared/model/prop_utils.py
@@ -1,18 +1,15 @@
 """Exposes utils for getting properties of the Model
 
-Attributes:
-    NESTED_MODEL_PREFIX (str): the prefix for fields with single nested models
-    NESTED_MODEL_LIST_FIELD_PREFIX (str): the prefix for fields with lists of nested models
-    NESTED_MODEL_TUPLE_FIELD_PREFIX (str): the prefix for fields with tuples of nested models
 """
 
+import re
 from typing import Type, Any
 
 from .base import AbstractModel
 
-NESTED_MODEL_PREFIX = "__"
-NESTED_MODEL_LIST_FIELD_PREFIX = "___"
-NESTED_MODEL_TUPLE_FIELD_PREFIX = "____"
+
+NESTED_MODEL_SEPARATOR = "_%&_"
+NESTED_MODEL_VALUE_REGEX = re.compile(f"^([\\w_]+{NESTED_MODEL_SEPARATOR}[\\w_]+)$")
 
 
 def get_redis_key(model: Type[AbstractModel], primary_key_value: Any):
@@ -38,7 +35,7 @@ def get_redis_key_prefix(model: Type[AbstractModel]):
         the prefix of the all the redis keys that are associated with this model
     """
     model_name = model.__name__.lower()
-    return f"{model_name}_%&_"
+    return f"{model_name}{NESTED_MODEL_SEPARATOR}"
 
 
 def get_redis_keys_regex(model: Type[AbstractModel]):
diff --git a/pydantic_redis/_shared/model/select_utils.py b/pydantic_redis/_shared/model/select_utils.py
index 33e0f842..114cf94c 100644
--- a/pydantic_redis/_shared/model/select_utils.py
+++ b/pydantic_redis/_shared/model/select_utils.py
@@ -2,22 +2,22 @@
 
 """
 
-from typing import List, Any, Type, Union, Awaitable, Optional
+from typing import List, Any, Type, Union, Awaitable, Optional, Dict, Tuple
 
 from pydantic_redis._shared.model.prop_utils import (
-    NESTED_MODEL_PREFIX,
-    NESTED_MODEL_LIST_FIELD_PREFIX,
-    NESTED_MODEL_TUPLE_FIELD_PREFIX,
     get_redis_keys_regex,
     get_redis_key_prefix,
     get_model_index_key,
 )
 
-
 from .base import AbstractModel
+from ..utils import groups_of_n
+
 
+RawRedisSelectData = List[Tuple[List[Any], List[Any]]]
 
-def get_select_fields(model: Type[AbstractModel], columns: List[str]) -> List[str]:
+
+def get_select_fields(model: Type[AbstractModel], columns: List[str] = ()) -> List[str]:
     """Gets the fields to be used for selecting HMAP fields in Redis.
 
     It replaces any fields in `columns` that correspond to nested records with their
@@ -30,29 +30,15 @@ def get_select_fields(model: Type[AbstractModel], columns: List[str]) -> List[st
     Returns:
         the fields for selecting, with nested fields being given appropriate prefixes.
     """
-    fields = []
-    nested_model_list_fields = model.get_nested_model_list_fields()
-    nested_model_tuple_fields = model.get_nested_model_tuple_fields()
-    nested_model_fields = model.get_nested_model_fields()
-
-    for col in columns:
-
-        if col in nested_model_fields:
-            col = f"{NESTED_MODEL_PREFIX}{col}"
-        elif col in nested_model_list_fields:
-            col = f"{NESTED_MODEL_LIST_FIELD_PREFIX}{col}"
-        elif col in nested_model_tuple_fields:
-            col = f"{NESTED_MODEL_TUPLE_FIELD_PREFIX}{col}"
-
-        fields.append(col)
-    return fields
+    typed_keys = model.get_field_typed_keys()
+    return [typed_keys.get(col, col) for col in columns]
 
 
 def select_all_fields_all_ids(
     model: Type[AbstractModel],
     skip: int = 0,
     limit: Optional[int] = None,
-) -> Union[List[List[Any]], Awaitable[List[List[Any]]]]:
+) -> Union[RawRedisSelectData, Awaitable[RawRedisSelectData]]:
     """Retrieves all records of the given model in the redis database.
 
     Args:
@@ -61,7 +47,7 @@ def select_all_fields_all_ids(
         limit: the maximum number of records to return. If None, limit is infinity.
 
     Returns:
-        the list of records from redis, each record being a flattened list of key-values.
+        list of tuple of [record, index-of-nested-models] with each record being a flattened list of key-values.
         In case we are using async, an Awaitable of that list is returned instead.
     """
     if isinstance(limit, int):
@@ -75,7 +61,7 @@ def select_all_fields_all_ids(
 
 def select_all_fields_some_ids(
     model: Type[AbstractModel], ids: List[str]
-) -> Union[List[List[Any]], Awaitable[List[List[Any]]]]:
+) -> Union[RawRedisSelectData, Awaitable[RawRedisSelectData]]:
     """Retrieves some records from redis.
 
     Args:
@@ -83,7 +69,7 @@ def select_all_fields_some_ids(
         ids: the list of primary keys of the records to be retrieved.
 
     Returns:
-        the list of records where each record is a flattened key-value list.
+        list of tuple of [record, index-of-nested-models] with each record is a flattened key-value list.
         In case we are using async, an Awaitable of that list is returned instead.
     """
     table_prefix = get_redis_key_prefix(model=model)
@@ -97,7 +83,7 @@ def select_some_fields_all_ids(
     fields: List[str],
     skip: int = 0,
     limit: Optional[int] = None,
-) -> Union[List[List[Any]], Awaitable[List[List[Any]]]]:
+) -> Union[RawRedisSelectData, Awaitable[RawRedisSelectData]]:
     """Retrieves records of model from redis, each as with a subset of the fields.
 
     Args:
@@ -107,7 +93,7 @@ def select_some_fields_all_ids(
         limit: the maximum number of records to return. If None, limit is infinity.
 
     Returns:
-        the list of records from redis, each record being a flattened list of key-values.
+        list of tuple of [record, index-of-nested-models] with each record being a flattened list of key-values.
         In case we are using async, an Awaitable of that list is returned instead.
     """
     columns = get_select_fields(model=model, columns=fields)
@@ -125,7 +111,7 @@ def select_some_fields_all_ids(
 
 def select_some_fields_some_ids(
     model: Type[AbstractModel], fields: List[str], ids: List[str]
-) -> Union[List[List[Any]], Awaitable[List[List[Any]]]]:
+) -> Union[RawRedisSelectData, Awaitable[RawRedisSelectData]]:
     """Retrieves some records of current model from redis, each as with a subset of the fields.
 
     Args:
@@ -134,7 +120,7 @@ def select_some_fields_some_ids(
         ids: the list of primary keys of the records to be retrieved.
 
     Returns:
-        the list of records from redis, each record being a flattened list of key-values.
+        list of tuple of [record, index-of-nested-models] with each record being a flattened list of key-values.
         In case we are using async, an Awaitable of that list is returned instead.
     """
     table_prefix = get_redis_key_prefix(model=model)
@@ -145,7 +131,7 @@ def select_some_fields_some_ids(
 
 
 def parse_select_response(
-    model: Type[AbstractModel], response: List[List], as_models: bool
+    model: Type[AbstractModel], response: RawRedisSelectData, as_models: bool
 ):
     """Casts a list of flattened key-value lists into a list of models or dicts.
 
@@ -167,17 +153,23 @@ def parse_select_response(
 
     if as_models:
         return [
-            model(**model.deserialize_partially(record))
-            for record in response
-            if record != []
+            model(
+                **model.deserialize_partially(record, index=_construct_index(raw_index))
+            )
+            for record, raw_index in response
+            if len(response) != 0
         ]
 
-    return [model.deserialize_partially(record) for record in response if record != []]
+    return [
+        model.deserialize_partially(record, index=_construct_index(raw_index))
+        for record, raw_index in response
+        if len(response) != 0
+    ]
 
 
 def _select_all_ids_all_fields_paginated(
     model: Type[AbstractModel], limit: int, skip: Optional[int]
-):
+) -> Union[RawRedisSelectData, Awaitable[RawRedisSelectData]]:
     """Retrieves a slice of all records of the given model in the redis database.
 
     Args:
@@ -186,7 +178,7 @@ def _select_all_ids_all_fields_paginated(
         limit: the maximum number of records to return. If None, limit is infinity.
 
     Returns:
-        the list of records from redis, each record being a flattened list of key-values.
+        list of tuple of [record, index-of-nested-models] with each record being a flattened list of key-values.
         In case we are using async, an Awaitable of that list is returned instead.
     """
     if skip is None:
@@ -199,7 +191,7 @@ def _select_all_ids_all_fields_paginated(
 
 def _select_some_fields_all_ids_paginated(
     model: Type[AbstractModel], columns: List[str], limit: int, skip: int
-):
+) -> Union[RawRedisSelectData, Awaitable[RawRedisSelectData]]:
     """Retrieves a slice of all records of model from redis, each as with a subset of the fields.
 
     Args:
@@ -209,7 +201,7 @@ def _select_some_fields_all_ids_paginated(
         limit: the maximum number of records to return. If None, limit is infinity.
 
     Returns:
-        the list of records from redis, each record being a flattened list of key-values.
+        list of tuple of [record, index-of-nested-models] with each record being a flattened list of key-values.
         In case we are using async, an Awaitable of that list is returned instead.
     """
     if skip is None:
@@ -218,3 +210,26 @@ def _select_some_fields_all_ids_paginated(
     args = [table_index_key, skip, limit, *columns]
     store = model.get_store()
     return store.paginated_select_some_fields_for_all_ids_script(args=args)
+
+
+def _construct_index(index_list: List[Any]) -> Dict[str, Any]:
+    """Constructs the index dict from the index list of nested models returned from redis
+
+    Args:
+        index_list: the flat list of the index of nested models, with key followed by [model, index] tuple
+            [key1, [model1_flat_list, index1_flat_list], key2, [model2_flat_list, index2_flat_list]...]
+
+    Returns:
+        the index as a dict
+    """
+    index = {}
+    for k, model_and_index in groups_of_n(index_list, 2):
+        model_as_list, index_as_list = model_and_index
+        model_index = _construct_index(index_as_list)
+        index[k] = {
+            # remove the dunders for nested model fields
+            key.lstrip("_"): model_index.get(value, value)
+            for key, value in groups_of_n(model_as_list, 2)
+        }
+
+    return index
diff --git a/pydantic_redis/_shared/utils.py b/pydantic_redis/_shared/utils.py
index 8fa95db7..7c1e87dd 100644
--- a/pydantic_redis/_shared/utils.py
+++ b/pydantic_redis/_shared/utils.py
@@ -3,10 +3,12 @@
 """
 
 import typing
-from typing import Any, Tuple, Optional, Union, Dict, Type, List
+from typing import Any, Tuple, Optional, Union, Dict, Type, List, Iterable, TypeVar
 
 import orjson
 
+T = TypeVar("T")
+
 
 def strip_leading(word: str, substring: str) -> str:
     """Strips the leading substring if it exists.
@@ -96,8 +98,12 @@ def from_str_or_bytes_to_any(value: Any, field_type: Type) -> Any:
     elif not isinstance(value, str):
         return value
 
-    # JSON parse all other values that are str
-    return orjson.loads(value)
+    try:
+        # JSON parse all other values that are str
+        return orjson.loads(value)
+    except orjson.JSONDecodeError:
+        # try to be as fault-tolerant as sanely possible
+        return value
 
 
 def from_any_to_valid_redis_type(value: Any) -> Union[str, bytes, List[Any]]:
@@ -155,3 +161,18 @@ def from_dict_to_key_value_list(data: Dict[str, Any]) -> List[Any]:
         parsed_list.append(v)
 
     return parsed_list
+
+
+def groups_of_n(items: Iterable[T], n: int) -> Iterable[Tuple[T, ...]]:
+    """Returns an iterable of tuples of size n from the given list of items
+
+    Note that it might ignore the last items if n does not fit nicely into the items list
+
+    Args:
+        items: the list of items from which to extract the tuples
+        n: the size of the tuples
+
+    Returns:
+        the iterable of tuples of n size from the list of items
+    """
+    return zip(*[iter(items)] * n)


Test patch:
diff --git a/test/conftest.py b/test/conftest.py
index 98356cdc..c7810186 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -1,6 +1,6 @@
 import socket
 from datetime import date
-from typing import Tuple, List, Optional
+from typing import Tuple, List, Optional, Dict
 
 import pytest
 import pytest_asyncio
@@ -56,6 +56,9 @@ class Library(syn.Model):
     lost: Optional[List[Book]] = None
     popular: Optional[Tuple[Book, Book]] = None
     new: Optional[Tuple[Book, Author, Book, int]] = None
+    list_of_tuples: Optional[List[Tuple[str, Book]]] = None
+    dict_of_models: Optional[Dict[str, Book]] = None
+    optional_nested: Optional[Book] = None
 
 
 class AsyncLibrary(asy.Model):
diff --git a/test/test_async_pydantic_redis.py b/test/test_async_pydantic_redis.py
index a90e19fa..a7cee471 100644
--- a/test/test_async_pydantic_redis.py
+++ b/test/test_async_pydantic_redis.py
@@ -5,7 +5,7 @@
 
 import pytest
 
-from pydantic_redis._shared.model.prop_utils import NESTED_MODEL_PREFIX  # noqa
+from pydantic_redis._shared.model.base import NESTED_MODEL_PREFIX  # noqa
 from pydantic_redis._shared.utils import strip_leading  # noqa
 from pydantic_redis.asyncio import Model, RedisConfig, Store
 from test.conftest import (
diff --git a/test/test_pydantic_redis.py b/test/test_pydantic_redis.py
index a3ddca41..5756c79a 100644
--- a/test/test_pydantic_redis.py
+++ b/test/test_pydantic_redis.py
@@ -6,8 +6,9 @@
 import pytest
 
 from pydantic_redis import Store
+from pydantic_redis._shared.model.prop_utils import NESTED_MODEL_SEPARATOR
 from pydantic_redis.config import RedisConfig  # noqa
-from pydantic_redis._shared.model.prop_utils import NESTED_MODEL_PREFIX  # noqa
+from pydantic_redis._shared.model.base import NESTED_MODEL_PREFIX  # noqa
 from pydantic_redis._shared.utils import strip_leading  # noqa
 from pydantic_redis.syncio.model import Model
 from test.conftest import (
@@ -199,6 +200,74 @@ def test_update_optional_nested_tuple_of_models(store: Store):
     assert got == expected
 
 
+@pytest.mark.parametrize("store", redis_store_fixture)
+def test_update_list_of_tuples_of_nested_models(store: Store):
+    list_of_tuples = [("some book", books[0]), ("book2", books[2])]
+    data = [
+        Library(
+            name="Babel Library", address="In a book", list_of_tuples=list_of_tuples
+        )
+    ]
+    Library.insert(data)
+    # the tuple of nested models is automatically inserted
+    got = sorted(Book.select(), key=lambda x: x.title)
+    expected_books = [book for _, book in list_of_tuples]
+    expected = sorted(expected_books, key=lambda x: x.title)
+    assert expected == got
+
+    got = sorted(Library.select(), key=lambda x: x.name)
+    expected = sorted(data, key=lambda x: x.name)
+    assert got == expected
+
+
+@pytest.mark.parametrize("store", redis_store_fixture)
+def test_update_dict_of_models(store: Store):
+    dict_of_models = {"some book": books[0], "book2": books[2]}
+    data = [
+        Library(
+            name="Babel Library", address="In a book", dict_of_models=dict_of_models
+        )
+    ]
+    Library.insert(data)
+    # the dict of nested models is automatically inserted
+    got = sorted(Book.select(), key=lambda x: x.title)
+    expected_books = [book for _, book in dict_of_models.items()]
+    expected = sorted(expected_books, key=lambda x: x.title)
+    assert expected == got
+
+    got = sorted(Library.select(), key=lambda x: x.name)
+    expected = sorted(data, key=lambda x: x.name)
+    assert got == expected
+
+
+@pytest.mark.parametrize("store", redis_store_fixture)
+def test_update_filled_optional_nested_model(store: Store):
+    data = [
+        Library(name="Babel Library", address="In a book", optional_nested=books[0])
+    ]
+    Library.insert(data)
+
+    got = sorted(Book.select(), key=lambda x: x.title)
+    assert [books[0]] == got
+
+    got = sorted(Library.select(), key=lambda x: x.name)
+    expected = sorted(data, key=lambda x: x.name)
+    assert got == expected
+
+
+@pytest.mark.parametrize("store", redis_store_fixture)
+def test_update_unfilled_optional_nested_model(store: Store):
+    data = [Library(name="Babel Library", address="In a book")]
+    Library.insert(data)
+
+    got = Book.select()
+    assert got is None
+
+    got = sorted(Library.select(), key=lambda x: x.name)
+    expected = sorted(data, key=lambda x: x.name)
+    assert got == expected
+
+
 @pytest.mark.parametrize("store", redis_store_fixture)
 def test_select_default(store: Store):
     """Selecting without arguments returns all the book models"""
@@ -432,7 +501,7 @@ def test_delete_multiple(store: Store):
 def __deserialize_book_data(raw_book_data: Dict[str, Any]) -> Book:
     """Deserializes the raw book data returning a book instance"""
     author_id = raw_book_data.pop(f"{NESTED_MODEL_PREFIX}author")
-    author_id = strip_leading(author_id, "author_%&_")
+    author_id = strip_leading(author_id, f"author{NESTED_MODEL_SEPARATOR}")
 
     data = Book.deserialize_partially(raw_book_data)