Update BaseExecutor and executors dir and tests for that

Author: sokolegg

Makefile

@@ -6,7 +6,8 @@ lint:
 	flake8 piper
 
 unit:
-	pytest -vs tests/import_tests.py
+	pytest -vs tests/import_test.py
+	pytest -vs tests/base_executor_test.py
 	pytest -vs tests/base_test.py
 	pytest -vs tests/envs_test.py::TestCompose
 	pytest -vs tests/envs_test.py::TestVenv

piper/base/executors/__init__.py

@@ -0,0 +1,3 @@
+from piper.base.executors._base_executor import BaseExecutor
+from piper.base.executors.http import HTTPExecutor
+from piper.base.executors.fastapi import FastAPIExecutor, FastAPITesseractExecutor

piper/base/executors/_base_executor.py

@@ -0,0 +1,56 @@
+from abc import abstractmethod
+
+from piper.envs import get_env, is_current_env, Env
+from piper.utils.logger_utils import logger
+
+
+class BaseExecutor:
+    """
+    This class is main executor which you need to inherit to work with piper normally.
+    This sync by default, but you can change to acync and use __call__ with await.
+    To create your child Executor just implement run for sync behavior or exec for async (set is_async)
+    or implement both run and exec
+
+    You can use prepared Executors like HTTPExecutor. Usually you don't need to control behavior for every environment.
+    However you can do that properly for your custom Executor :
+
+    class YourCustomExecutor(BaseExecutor):
+        def run():
+            x + x
+        def docker_run():
+            ... # your custom logic for docker env
+        def compose_run():
+            ... # your custom logic for compose env
+        def custom_env_run():
+            ... # for you own env
+    """
+
+    is_async: bool = False
+
+    @abstractmethod
+    def run(self, *args, **kwargs):
+        raise NotImplementedError(f"run method not implemented in Executor {self}")
+
+    @abstractmethod
+    async def exec(self, *args, **kwargs):
+        raise NotImplementedError(f"exec method not implemented in Executor {self}")
+
+    def env_run(self, env: Env, *args, **kwargs):
+        if is_current_env():
+            return self.run(*args, **kwargs)
+        else:
+            env_run_name = f"{env.name}_run"
+            return getattr(self, env_run_name)(*args, **kwargs)
+
+    async def env_exec(self, env: Env, *args, **kwargs):
+        if is_current_env():
+            return await self.exec(*args, **kwargs)
+        else:
+            env_run_name = f"{env.name}_exec"
+            return await getattr(self, env_run_name)(*args, **kwargs)
+
+    def __call__(self, *args, **kwargs):
+        if self.is_async:
+            return self.env_exec(get_env(), *args, **kwargs)
+        else:
+            return self.env_run(get_env(), *args, **kwargs)

piper/base/executors.py renamed to piper/base/executors/fastapi.py

@@ -3,9 +3,12 @@
                                       render_fast_api_tsrct_backend)
 from piper.base.docker import PythonImage
 from piper.configurations import get_configuration
-from piper.envs import get_env, is_current_env, is_docker_env
+from piper.envs import get_env, is_current_env, is_docker_env, Env
 from piper.utils import docker_utils
+from piper.utils.logger_utils import logger
+from piper.base.executors import HTTPExecutor
 
+import asyncio
 import inspect
 import sys
 import time
@@ -18,38 +21,6 @@
 import requests
 from pydantic import BaseModel  # , BytesObject, ListOfStringsObject
 
-from piper.utils.logger_utils import logger
-
-
-class BaseExecutor:
-    pass
-
-
-class LocalExecutor:
-    pass
-
-
-def is_known(obj):
-    basic = obj.__class__.__name__ in {'dict', 'list', 'tuple', 'str', 'int', 'float', 'bool'}
-    models = isinstance(obj, (BaseModel,))
-    return basic or models
-
-
-def prepare(obj):
-    if isinstance(obj, (BaseModel,)):
-        return obj.dict()
-    return obj
-
-
-def inputs_to_dict(*args, **kwargs):
-    from_args = {}
-    for arg in args:
-        if is_known(arg):
-            from_args.update(prepare(arg))
-    from_kwargs = {k: prepare(v) for k, v in kwargs.items() if is_known(v)}
-    from_args.update(from_kwargs)
-    return from_args
-
 
 def add_packages_to_install(packages_list):
     row = f'RUN apt install -y {" ".join(packages_list)} \n'
@@ -60,32 +31,6 @@ def add_row(row):
     return f'{row} \n'
 
 
-class HTTPExecutor(BaseExecutor):
-
-    def __init__(self, host: str, port: int, base_handler: str):
-        self.host = host
-        self.port = port
-
-    @abstractmethod
-    async def run(self, *args, **kwargs):
-        pass
-
-    async def __call__(self, *args, **kwargs):
-        logger.info(f'get_env() {get_env()}')
-        logger.info(f'is_current_env() {is_current_env()}')
-        if is_current_env():
-            return await self.run(*args, **kwargs)
-        else:
-            function = "run"
-            request_dict = inputs_to_dict(*args, **kwargs)
-            logger.info(f'request_dict is {request_dict}')
-            async with aiohttp.ClientSession() as session:
-                url = f'http://{self.host}:{self.port}/{function}'
-                logger.info(f'run function with url {url} and data {request_dict}')
-                async with session.post(url, json=request_dict) as resp:
-                    return await resp.json()
-
-
 def copy_piper(path: str):
     cfg = get_configuration()
     copy_tree(cfg.piper_path, f"{path}piper")
@@ -162,7 +107,8 @@ class FastAPIExecutor(HTTPExecutor):
     def __init__(self, port: int = 8080, **service_kwargs):
         self.container = None
         self.image_tag = 'piper:latest'
-        self.container_name = "piper_FastAPI"
+        self.id = hash(self)
+        self.container_name = f"piper_FastAPI_{self.id}"
 
         if is_docker_env():
             docker_client = docker.DockerClient(base_url='unix://var/run/docker.sock')
@@ -194,6 +140,14 @@ def __init__(self, port: int = 8080, **service_kwargs):
 
         super().__init__('localhost', port, self.base_handler)
 
+    async def aio_call(self, *args, **kwargs):
+        return await super().__call__(*args, ** kwargs)
+
+    def __call__(self, *args, **kwargs):
+        loop = asyncio.get_event_loop()
+        result = loop.run_until_complete(self.aio_call(*args, **kwargs))
+        return result
+
     def rm_container(self):
         if self.container:
             self.container.remove(force=True)

piper/base/executors/http.py

@@ -0,0 +1,56 @@
+from abc import abstractmethod
+
+import aiohttp
+
+from piper.envs import get_env, is_current_env
+from piper.utils.logger_utils import logger
+from piper.base.executors import BaseExecutor
+from pydantic import BaseModel
+
+
+def is_known(obj):
+    basic = obj.__class__.__name__ in {'dict', 'list', 'tuple', 'str', 'int', 'float', 'bool'}
+    models = isinstance(obj, (BaseModel,))
+    return basic or models
+
+
+def prepare(obj):
+    if isinstance(obj, (BaseModel,)):
+        return obj.dict()
+    return obj
+
+
+def inputs_to_dict(*args, **kwargs):
+    from_args = {}
+    for arg in args:
+        if is_known(arg):
+            from_args.update(prepare(arg))
+    from_kwargs = {k: prepare(v) for k, v in kwargs.items() if is_known(v)}
+    from_args.update(from_kwargs)
+    return from_args
+
+
+class HTTPExecutor(BaseExecutor):
+
+    def __init__(self, host: str, port: int, base_handler: str):
+        self.host = host
+        self.port = port
+
+    @abstractmethod
+    async def run(self, *args, **kwargs):
+        pass
+
+    async def __call__(self, *args, **kwargs):
+        logger.info(f'get_env() {get_env()}')
+        logger.info(f'is_current_env() {is_current_env()}')
+        if is_current_env():
+            return await self.run(*args, **kwargs)
+        else:
+            function = "run"
+            request_dict = inputs_to_dict(*args, **kwargs)
+            logger.info(f'request_dict is {request_dict}')
+            async with aiohttp.ClientSession() as session:
+                url = f'http://{self.host}:{self.port}/{function}'
+                logger.info(f'run function with url {url} and data {request_dict}')
+                async with session.post(url, json=request_dict) as resp:
+                    return await resp.json()

piper/configurations.py

@@ -5,13 +5,13 @@ class Configuration:
     path: str = f"./applications/piper_project_{time.time_ns()}/"
     test_path: str = f"./applications/piper_project_{time.time_ns()}/"
     piper_path: str = "piper"
-    default_env: str = "docker"
+    default_env: str = "compose"
     docker_app_port: int = 8788
 
     name_venv: str = "venv_test"
     number: int = 10
 
-    env: str = 'compose'
+    env: str = None
     ignore_import_errors: bool = True
     safe_import_activated: bool = False
 

piper/envs/__init__.py

@@ -16,7 +16,7 @@ def init_default_env():
     elif cfg.default_env == "virtualenv":
         set_env(VirtualEnv())
     elif cfg.default_env == "compose":
-        set_env(VirtualEnv())
+        set_env(ComposeEnv())
     else:
         set_env(CurrentEnv())
 
@@ -32,69 +32,66 @@ def set_env(env):
     cfg.env = env
 
 
-class DockerEnv:
+class Env:
+    name = "no_env"
 
-    def __init__(self):
-        pass
+    _subclasses = []
 
     def __enter__(self):
-        print("Entering DockerEnv")
+        logger.info(f"Entering Env: {self.__class__.__name__}")
         self._old_environment = get_env()
         set_env(self)
 
     def __exit__(self, *args, **kws):
-        print("Exiting DockerEnv")
+        logger.info(f"Exit Env: {self.__class__.__name__}")
         set_env(self._old_environment)
 
+    @classmethod
+    def get_all_envs(cls):
+        return list(cls._subclasses)
 
-class CurrentEnv:
+    def __init_subclass__(cls):
+        Env._subclasses.append(cls)
+
+
+class DockerEnv(Env):
+    name = "docker"
 
     def __init__(self):
         pass
 
-    def __enter__(self):
-        print("Entering CurrentEnv")
-        self._old_environment = get_env()
-        set_env(self)
 
-    def __exit__(self, *args, **kws):
-        print("Exiting CurrentEnv")
-        set_env(self._old_environment)
+class CurrentEnv(Env):
+    name = "current_env"
+
+    def __init__(self):
+        pass
 
 
-class VirtualEnv:
+class VirtualEnv(Env):
+    name = "virtualenv"
 
     def __init__(self):
         self.__resource = VirtualEnvExecutor()
 
     def __enter__(self):
-        logger.info("Entering VirtualEnv")
-        self._old_environment = get_env()
-        set_env(self)
-        # TODO update work with return resource
+        super().__enter__()
         return self.__resource
 
-    def __exit__(self, *args, **kws):
-        logger.info("Exiting VirtualEnv")
-        set_env(self._old_environment)
-
 
-class ComposeEnv:
+class ComposeEnv(Env):
+    name = "compose"
 
     def __init__(self):
         self.__resource = ComposeExecutor()
 
     def __enter__(self):
-        logger.info("Entering ComposeEnv")
-        self._old_environment = get_env()
-        set_env(self)
-        # TODO update work with return resource
+        super().__enter__()
         return self.__resource
 
     def __exit__(self, *args, **kws):
-        logger.info("Exiting ComposeEnv")
+        super().__exit__(* args, ** kws)
         # self.__resource.stop_compose()
-        set_env(self._old_environment)
 
 
 def is_current_env():