Migration to async DB drivers

.github/workflows/test_full.yml

@@ -44,5 +44,5 @@ jobs:
           if [ "$RUNNER_OS" == "Linux" ]; then
               source <(cat .env_file | sed 's/=/=/' | sed 's/^/export /')
           fi
-          pytest --tb=long -vv --cache-clear --cov=dff tests/
+          pytest -rs --tb=long -vv --cache-clear --cov=dff tests/
         shell: bash

README.md

@@ -113,7 +113,7 @@ from dff.script import Context, Actor
 from dff.context_storages import SQLContextStorage
 from .script import some_df_script
 
-db = SQLContextStorage("postgresql://user:password@host:port/dbname")
+db = SQLContextStorage("postgresql+asyncpg://user:password@host:port/dbname")
 
 actor = Actor(some_df_script, start_label=("root", "start"), fallback_label=("root", "fallback"))
 

dff/context_storages/__init__.py

@@ -1,9 +1,14 @@
 # -*- coding: utf-8 -*-
 # flake8: noqa: F401
+# fmt: off
 
-from .database import DBAbstractContextStorage, DBContextStorage, threadsafe_method, context_storage_factory
-from .json import JSONContextStorage
-from .pickle import PickleContextStorage
+import nest_asyncio
+
+nest_asyncio.apply()
+
+from .database import DBContextStorage, threadsafe_method, context_storage_factory
+from .json import JSONContextStorage, json_available
+from .pickle import PickleContextStorage, pickle_available
 from .sql import SQLContextStorage, postgres_available, mysql_available, sqlite_available, sqlalchemy_available
 from .ydb import YDBContextStorage, ydb_available
 from .redis import RedisContextStorage, redis_available

dff/context_storages/database.py

@@ -2,85 +2,176 @@
 Database
 --------
 Base module. Provided classes:
-    - Abstract context storage interface :py:class:`.DBAbstractContextStorage`.
-    - An intermediate class to inherit from: :py:class:`.DBContextStorage`
+    - Abstract context storage interface :py:class:`.DBContextStorage`.
 """
+import asyncio
 import importlib
 import threading
 from functools import wraps
 from abc import ABC, abstractmethod
-from typing import Any, Callable
+from typing import Callable, Hashable, Optional
 
 from .protocol import PROTOCOLS
+from ..script import Context
 
 
-class DBAbstractContextStorage(ABC):
+class DBContextStorage(ABC):
     """
     An abstract interface for `dff` DB context storages.
     It includes the most essential methods of the python `dict` class.
     Can not be instantiated.
+
+    Parameters
+    ----------
+    :param path:
+        | Parameter `path` should be set with the URI of the database.
+        | It includes a prefix and the required connection credentials.
+        | Example: postgresql+asyncpg://user:password@host:port/database
+        | In the case of classes that save data to hard drive instead of external databases
+        | you need to specify the location of the file, like you do in sqlite.
+        | Keep in mind that in Windows you will have to use double backslashes '\\'
+        | instead of forward slashes '/' when defining the file path.
+    :type path: str
+    :param full_path: Full path to access the context storage, as it was provided by user.
+    :type full_path: str
+    :param _lock: Threading for methods that require single thread access.
+
     """
 
-    def __init__(self) -> None:
-        pass
+    def __init__(self, path: str):
+        _, _, file_path = path.partition("://")
+        self.full_path = path
+        self.path = file_path
+        self._lock = threading.Lock()
 
-    @abstractmethod
-    def __getitem__(self, key: str) -> Any:
-        raise NotImplementedError
+    def __getitem__(self, key: Hashable) -> Context:
+        """
+        Synchronous method for accessing stored Context.
+        :param key: Hashable key used to store Context instance.
+        :type key: Hashable
+        :returns: The stored context, associated with the given key.
+        """
+        return asyncio.run(self.get_item_async(key))
 
     @abstractmethod
-    def __setitem__(self, key: str, value: dict) -> None:
+    async def get_item_async(self, key: Hashable) -> Context:
+        """
+        Asynchronous method for accessing stored Context.
+        :param key: Hashable key used to store Context instance.
+        :type key: Hashable
+        :returns: The stored context, associated with the given key.
+        """
         raise NotImplementedError
 
-    @abstractmethod
-    def __delitem__(self, key: str) -> None:
-        raise NotImplementedError
+    def __setitem__(self, key: Hashable, value: Context):
+        """
+        Synchronous method for storing Context.
+        :param key: Hashable key used to store Context instance.
+        :type key: Hashable
+        :param value: Context to store.
+        :type value: Context
+        """
+        return asyncio.run(self.set_item_async(key, value))
 
     @abstractmethod
-    def __contains__(self, key: str) -> bool:
+    async def set_item_async(self, key: Hashable, value: Context):
+        """
+        Asynchronous method for storing Context.
+        :param key: Hashable key used to store Context instance.
+        :type key: Hashable
+        :param value: Context to store.
+        :type value: Context
+        """
         raise NotImplementedError
 
-    @abstractmethod
-    def __len__(self) -> int:
-        raise NotImplementedError
+    def __delitem__(self, key: Hashable):
+        """
+        Synchronous method for removing stored Context.
+        :param key: Hashable key used to identify Context instance for deletion.
+        :type key: Hashable
+        """
+        return asyncio.run(self.del_item_async(key))
 
     @abstractmethod
-    def get(self, item) -> Any:
+    async def del_item_async(self, key: Hashable):
+        """
+        Asynchronous method for removing stored Context.
+        :param key: Hashable key used to identify Context instance for deletion.
+        :type key: Hashable
+        """
         raise NotImplementedError
 
+    def __contains__(self, key: Hashable) -> bool:
+        """
+        Synchronous method for finding whether any Context is stored with given key.
+        :param key: Hashable key used to check if Context instance is stored.
+        :type key: Hashable
+        :returns: True if there is Context accessible by given key, False otherwise.
+        """
+        return asyncio.run(self.contains_async(key))
+
     @abstractmethod
-    def clear(self) -> None:
+    async def contains_async(self, key: Hashable) -> bool:
+        """
+        Asynchronous method for finding whether any Context is stored with given key.
+        :param key: Hashable key used to check if Context instance is stored.
+        :type key: Hashable
+        :returns: True if there is Context accessible by given key, False otherwise.
+        """
         raise NotImplementedError
 
+    def __len__(self) -> int:
+        """
+        Synchronous method for retrieving number of stored Contexts.
+        :returns: The number of stored Contexts.
+        """
+        return asyncio.run(self.len_async())
 
-class DBContextStorage(DBAbstractContextStorage):
-    """
-    An intermediate class between the abstract context storage interface,
-    :py:class:`.DBAbstractContextStorage`, and concrete implementations.
-
-    :param path: Parameter `path` should be set with the URI of the database.
-        It includes a prefix and the required connection credentials.
-        Example: postgresql://user:password@host:port/database
-        In the case of classes that save data to hard drive instead of external databases
-        you need to specify the location of the file, like you do in sqlite.
-        Keep in mind that in Windows you will have to use double backslashes '\\'
-        instead of forward slashes '/' when defining the file path.
-    :type path: str
-
-    """
-
-    def __init__(self, path: str):
-        _, _, file_path = path.partition("://")
-        self.full_path = path
-        self.path = file_path
-        self._lock = threading.Lock()
+    @abstractmethod
+    async def len_async(self) -> int:
+        """
+        Asynchronous method for retrieving number of stored Contexts.
+        :returns: The number of stored Contexts.
+        """
+        raise NotImplementedError
 
-    def get(self, key: str, default=None) -> Any:
+    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.
+        :param key: Hashable key used to store Context instance.
+        :type key: Hashable
+        :param default: Optional default value to be returned if no Context is found.
+        :type key: Optional[Context]
+        :returns: The stored context, associated with the given key or default value.
+        """
+        return asyncio.run(self.get_async(key, default))
+
+    async def get_async(self, key: Hashable, default: Optional[Context] = None) -> Context:
+        """
+        Asynchronous method for accessing stored Context, returning default if no Context is stored with the given key.
+        :param key: Hashable key used to store Context instance.
+        :type key: Hashable
+        :param default: Optional default value to be returned if no Context is found.
+        :type key: Optional[Context]
+        :returns: The stored context, associated with the given key or default value.
+        """
         try:
-            value = self.__getitem__(key)
+            return await self.get_item_async(str(key))
         except KeyError:
-            value = default
-        return value
+            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):
@@ -96,24 +187,23 @@ def _synchronized(self, *args, **kwargs):
     return _synchronized
 
 
-def context_storage_factory(path: str, **kwargs):
+def context_storage_factory(path: str, **kwargs) -> DBContextStorage:
     """
     Use context_storage_factory to lazy import context storage types and instantiate them.
     The function takes a database connection URI or its equivalent. It should be prefixed with database name,
     followed by the symbol triplet '://'.
     Then, you should list the connection parameters like this: user:password@host:port/database
     The whole URI will then look like this:
-
-        - shelve://path_to_the_file/file_name
-        - json://path_to_the_file/file_name
-        - pickle://path_to_the_file/file_name
-        - sqlite://path_to_the_file/file_name
-        - redis://:pass@localhost:6379/0
-        - mongodb://admin:pass@localhost:27017/admin
-        - mysql+pymysql://root:pass@localhost:3307/test
-        - postgresql://postgres:pass@localhost:5432/test
-        - grpc://localhost:2136/local
-        - grpcs://localhost:2135/local
+    - shelve://path_to_the_file/file_name
+    - json://path_to_the_file/file_name
+    - pickle://path_to_the_file/file_name
+    - sqlite+aiosqlite://path_to_the_file/file_name
+    - redis://:pass@localhost:6378/0
+    - mongodb://admin:pass@localhost:27016/admin
+    - mysql+asyncmy://root:pass@localhost:3306/test
+    - postgresql+asyncpg://postgres:pass@localhost:5430/test
+    - grpc://localhost:2134/local
+    - grpcs://localhost:2134/local
 
     For context storages that write to local files, the function expects a file path instead of connection params:
     json://file.json