Partial context updates

dff/context_storages/__init__.py

@@ -10,3 +10,4 @@
 from .mongo import MongoContextStorage, mongo_available
 from .shelve import ShelveContextStorage
 from .protocol import PROTOCOLS, get_protocol_install_suggestion
+from .update_scheme import UpdateScheme

dff/context_storages/database.py

@@ -12,8 +12,10 @@
 import threading
 from functools import wraps
 from abc import ABC, abstractmethod
+from inspect import signature
 from typing import Callable, Hashable, Optional
 
+from .update_scheme import UpdateScheme
 from .protocol import PROTOCOLS
 from ..script import Context
 
@@ -34,14 +36,19 @@ class DBContextStorage(ABC):
 
     """
 
-    def __init__(self, path: str):
+    def __init__(self, path: str, update_scheme: Optional[UpdateScheme] = None):
         _, _, file_path = path.partition("://")
         self.full_path = path
         """Full path to access the context storage, as it was provided by user."""
         self.path = file_path
         """`full_path` without a prefix defining db used"""
         self._lock = threading.Lock()
         """Threading for methods that require single thread access."""
+        self.hash_storage = dict()
+        self.set_update_scheme(update_scheme)
+
+    def set_update_scheme(self, update_scheme: Optional[UpdateScheme]):
+        self.update_scheme = update_scheme if update_scheme else UpdateScheme()
 
     def __getitem__(self, key: Hashable) -> Context:
         """
@@ -134,6 +141,19 @@ async def len_async(self) -> int:
         """
         raise NotImplementedError
 
+    def clear(self):
+        """
+        Synchronous method for clearing context storage, removing all the stored Contexts.
+        """
+        return asyncio.run(self.clear_async())
+
+    @abstractmethod
+    async def clear_async(self):
+        """
+        Asynchronous method for clearing context storage, removing all the stored Contexts.
+        """
+        raise NotImplementedError
+
     def get(self, key: Hashable, default: Optional[Context] = None) -> Context:
         """
         Synchronous method for accessing stored Context, returning default if no Context is stored with the given key.
@@ -157,19 +177,6 @@ async def get_async(self, key: Hashable, default: Optional[Context] = None) -> C
         except KeyError:
             return default
 
-    def clear(self):
-        """
-        Synchronous method for clearing context storage, removing all the stored Contexts.
-        """
-        return asyncio.run(self.clear_async())
-
-    @abstractmethod
-    async def clear_async(self):
-        """
-        Asynchronous method for clearing context storage, removing all the stored Contexts.
-        """
-        raise NotImplementedError
-
 
 def threadsafe_method(func: Callable):
     """
@@ -184,6 +191,18 @@ def _synchronized(self, *args, **kwargs):
     return _synchronized
 
 
+def auto_stringify_hashable_key(key_name: str = "key"):
+    def auto_stringify(func: Callable):
+        all_keys = signature(func).parameters.keys()
+
+        async def stringify_arg(*args, **kwargs):
+            return await func(*[str(arg) if name == key_name else arg for arg, name in zip(args, all_keys)], **kwargs)
+
+        return stringify_arg
+
+    return auto_stringify
+
+
 def context_storage_factory(path: str, **kwargs) -> DBContextStorage:
     """
     Use context_storage_factory to lazy import context storage types and instantiate them.

dff/context_storages/json.py

@@ -6,7 +6,11 @@
 store and retrieve context data.
 """
 import asyncio
-from typing import Hashable
+from typing import Hashable, Union, List, Any, Dict, Tuple, Optional
+
+from pydantic import BaseModel, Extra, root_validator
+
+from .update_scheme import UpdateScheme, FieldRule
 
 try:
     import aiofiles
@@ -15,18 +19,17 @@
     json_available = True
 except ImportError:
     json_available = False
+    aiofiles = None
 
-from pydantic import BaseModel, Extra, root_validator
-
-from .database import DBContextStorage, threadsafe_method
+from .database import DBContextStorage, threadsafe_method, auto_stringify_hashable_key
 from dff.script import Context
 
 
 class SerializableStorage(BaseModel, extra=Extra.allow):
     @root_validator
     def validate_any(cls, vals):
-        for key, value in vals.items():
-            vals[key] = Context.cast(value)
+        for key, values in vals.items():
+            vals[key] = [None if value is None else Context.cast(value) for value in values]
         return vals
 
 
@@ -41,33 +44,56 @@ def __init__(self, path: str):
         DBContextStorage.__init__(self, path)
         asyncio.run(self._load())
 
-    @threadsafe_method
-    async def len_async(self) -> int:
-        return len(self.storage.__dict__)
+    def set_update_scheme(self, scheme: UpdateScheme):
+        super().set_update_scheme(scheme)
+        self.update_scheme.mark_db_not_persistent()
+        self.update_scheme.id.on_write = FieldRule.UPDATE
 
     @threadsafe_method
-    async def set_item_async(self, key: Hashable, value: Context):
-        self.storage.__dict__.__setitem__(str(key), value)
-        await self._save()
+    @auto_stringify_hashable_key()
+    async def get_item_async(self, key: Union[Hashable, str]) -> Context:
+        await self._load()
+        fields, int_id = await self._read_keys(key)
+        if int_id is None:
+            raise KeyError(f"No entry for key {key}.")
+        context, hashes = await self.update_scheme.read_context(fields, self._read_ctx, key, int_id)
+        self.hash_storage[key] = hashes
+        return context
 
     @threadsafe_method
-    async def get_item_async(self, key: Hashable) -> Context:
-        await self._load()
-        return Context.cast(self.storage.__dict__.__getitem__(str(key)))
+    @auto_stringify_hashable_key()
+    async def set_item_async(self, key: Union[Hashable, str], value: Context):
+        fields, _ = await self._read_keys(key)
+        value_hash = self.hash_storage.get(key, None)
+        await self.update_scheme.write_context(value, value_hash, fields, self._write_ctx, key)
+        await self._save()
 
     @threadsafe_method
-    async def del_item_async(self, key: Hashable):
-        self.storage.__dict__.__delitem__(str(key))
+    @auto_stringify_hashable_key()
+    async def del_item_async(self, key: Union[Hashable, str]):
+        container = self.storage.__dict__.get(key, list())
+        container.append(None)
+        self.storage.__dict__[key] = container
         await self._save()
 
     @threadsafe_method
-    async def contains_async(self, key: Hashable) -> bool:
+    @auto_stringify_hashable_key()
+    async def contains_async(self, key: Union[Hashable, str]) -> bool:
         await self._load()
-        return self.storage.__dict__.__contains__(str(key))
+        if key in self.storage.__dict__:
+            container = self.storage.__dict__.get(key, list())
+            if len(container) != 0:
+                return container[-1] is not None
+        return False
+
+    @threadsafe_method
+    async def len_async(self) -> int:
+        return len(self.storage.__dict__)
 
     @threadsafe_method
     async def clear_async(self):
-        self.storage.__dict__.clear()
+        for key in self.storage.__dict__.keys():
+            await self.del_item_async(key)
         await self._save()
 
     async def _save(self):
@@ -81,3 +107,36 @@ async def _load(self):
         else:
             async with aiofiles.open(self.path, "r", encoding="utf-8") as file_stream:
                 self.storage = SerializableStorage.parse_raw(await file_stream.read())
+
+    async def _read_keys(self, ext_id: str) -> Tuple[Dict[str, List[str]], Optional[str]]:
+        key_dict = dict()
+        container = self.storage.__dict__.get(ext_id, list())
+        if len(container) == 0:
+            return key_dict, None
+        container_dict = container[-1].dict() if container[-1] is not None else dict()
+        for field in [key for key, value in container_dict.items() if isinstance(value, dict)]:
+            key_dict[field] = list(container_dict.get(field, dict()).keys())
+        return key_dict, container_dict.get(self.update_scheme.id.name, None)
+
+    async def _read_ctx(self, outlook: Dict[str, Union[bool, Dict[Hashable, bool]]], _: str, ext_id: str) -> Dict:
+        result_dict = dict()
+        context = self.storage.__dict__[ext_id][-1].dict()
+        for field in [field for field, value in outlook.items() if isinstance(value, dict) and len(value) > 0]:
+            for key in [key for key, value in outlook[field].items() if value]:
+                value = context.get(field, dict()).get(key, None)
+                if value is not None:
+                    if field not in result_dict:
+                        result_dict[field] = dict()
+                    result_dict[field][key] = value
+        for field in [field for field, value in outlook.items() if isinstance(value, bool) and value]:
+            value = context.get(field, None)
+            if value is not None:
+                result_dict[field] = value
+        return result_dict
+
+    async def _write_ctx(self, data: Dict[str, Any], _: str, ext_id: str):
+        container = self.storage.__dict__.setdefault(ext_id, list())
+        if len(container) > 0:
+            container[-1] = Context.cast({**container[-1].dict(), **data})
+        else:
+            container.append(Context.cast(data))