using pydantic

passes unit tests now

Author: sanderegg

services/sidecar/src/simcore_service_sidecar/cli.py

@@ -1,4 +1,5 @@
 import logging
+from typing import List
 
 import click
 
@@ -12,9 +13,8 @@
 @click.option("--user_id", default="0", help="The user ID")
 @click.option("--project_id", default="0", help="The project ID")
 @click.option("--node_id", default=None, help="The node ID or nothing")
-def main(job_id: str, user_id: str, project_id: str, node_id: str):
+def main(job_id: str, user_id: str, project_id: str, node_id: str) -> List[str]:
     from simcore_service_sidecar.core import SIDECAR
-    import pdb; pdb.set_trace()
     log.info(
         "STARTING task processing for user %s, project %s, node %s",
         user_id,
@@ -31,3 +31,5 @@ def main(job_id: str, user_id: str, project_id: str, node_id: str):
         project_id,
         node_id,
     )
+
+    return next_task_nodes

services/sidecar/src/simcore_service_sidecar/core.py

@@ -6,13 +6,12 @@
 from contextlib import contextmanager
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, List, Union, Any
+from typing import Any, Dict
 
-import aio_pika
-import aiofiles
-import attr
+# import aiofiles
 import docker
 from celery.utils.log import get_task_logger
+from pydantic import BaseModel
 from sqlalchemy import and_, exc
 
 from servicelib.utils import logged_gather
@@ -26,7 +25,7 @@
 from . import config
 from .rabbitmq import RabbitMQ
 from .utils import (DbSettings, DockerSettings, ExecutorSettings, S3Settings,
-                    find_entry_point, is_node_ready, safe_channel)
+                    find_entry_point, is_node_ready)
 
 log = get_task_logger(__name__)
 log.setLevel(config.SIDECAR_LOGLEVEL)
@@ -48,8 +47,7 @@ def session_scope(session_factory):
     finally:
         session.close()
 
-@attr.s(auto_attribs=True)
-class Sidecar:  # pylint: disable=too-many-instance-attributes
+class Sidecar(BaseModel):
     _rabbit_mq: RabbitMQ
     _docker: DockerSettings = DockerSettings()
     _s3: S3Settings = S3Settings()
@@ -153,34 +151,6 @@ async def _pull_image(self):
             log.exception(msg)
             raise docker.errors.APIError(msg)
 
-    async def _post_log(
-        self, channel: pika.channel.Channel, msg: Union[str, List[str]]
-    ):
-        log_data = {
-            "Channel": "Log",
-            "Node": self._task.node_id,
-            "user_id": self._user_id,
-            "project_id": self._task.project_id,
-            "Messages": msg if isinstance(msg, list) else [msg],
-        }
-        log_body = json.dumps(log_data)
-        channel.basic_publish(
-            exchange=self._pika.log_channel, routing_key="", body=log_body
-        )
-
-    async def _post_progress(self, channel, progress):
-        prog_data = {
-            "Channel": "Progress",
-            "Node": self._task.node_id,
-            "user_id": self._user_id,
-            "project_id": self._task.project_id,
-            "Progress": progress,
-        }
-        prog_body = json.dumps(prog_data)
-        channel.basic_publish(
-            exchange=self._pika.progress_channel, routing_key="", body=prog_body
-        )
-
     async def log_file_processor(self, log_file: Path) -> None:
         """checks both container logs and the log_file if any
         """
@@ -208,7 +178,6 @@ async def log_file_processor(self, log_file: Path) -> None:
 
 
         # try:
-        #     import pdb; pdb.set_trace()
         #     TIME_BETWEEN_LOGS_S: int = 2
         #     time_logs_sent = time.monotonic()
         #     accumulated_logs = []
@@ -508,25 +477,17 @@ async def run(self):
             self._task.node_id,
             self._task.internal_id,
         )
-        # NOTE: the rabbit has a timeout of 60seconds so blocking this channel for more is a no go.
-
-        with safe_channel(self._pika) as (channel, _):
-            await self._post_log(channel, msg="Preprocessing start...")
-
+        await self._rabbit_mq.post_log_message("Preprocessing start...")
         await self.preprocess()
-
-        with safe_channel(self._pika) as (channel, _):
-            await self._post_log(channel, msg="...preprocessing end")
-            await self._post_log(channel, msg="Processing start...")
+        await self._rabbit_mq.post_log_message("...preprocessing end")
+        
+        await self._rabbit_mq.post_log_message("Processing start...")
         await self.process()
+        await self._rabbit_mq.post_log_message("...processing end")
 
-        with safe_channel(self._pika) as (channel, _):
-            await self._post_log(channel, msg="...processing end")
-            await self._post_log(channel, msg="Postprocessing start...")
+        await self._rabbit_mq.post_log_message("Postprocessing start...")
         await self.postprocess()
-
-        with safe_channel(self._pika) as (channel, _):
-            await self._post_log(channel, msg="...postprocessing end")
+        await self._rabbit_mq.post_log_message("...postprocessing end")
 
         log.debug(
             "Running Pipeline DONE %s:node %s:internal id %s from container",
@@ -617,7 +578,7 @@ async def inspect(self, job_request_id: int, user_id: str, project_id: str, node
                 return next_task_nodes
 
             # the task is ready!
-            task.job_id = celery_task.request.id
+            task.job_id = job_request_id
             _session.add(task)
             _session.commit()
 
@@ -632,7 +593,7 @@ async def inspect(self, job_request_id: int, user_id: str, project_id: str, node
                 .one()
             )
 
-            if task.job_id != celery_task.request.id:
+            if task.job_id != job_request_id:
                 # somebody else was faster
                 return next_task_nodes
             task.state = RUNNING

services/sidecar/src/simcore_service_sidecar/rabbitmq.py

@@ -1,46 +1,89 @@
+import json
 import logging
+from typing import Dict, List, Union
 
 import aio_pika
-import attr
 import tenacity
+from pydantic import BaseModel
 
 from servicelib.rabbitmq_utils import RabbitMQRetryPolicyUponInitialization
-from simcore_sdk.config.rabbit import Config, eval_broker
+from simcore_sdk.config.rabbit import Config as RabbitConfig
+from simcore_sdk.config.rabbit import eval_broker
 
 log = logging.getLogger(__file__)
 
-@attr.s(auto_attribs=True)
-class RabbitMQ:
-    _config: Config = Config()
+
+class RabbitMQ(BaseModel):
+    _config: RabbitConfig = RabbitConfig()
     _connection: aio_pika.RobustConnection = None
     _channel: aio_pika.Channel = None
-    logs_exchange: aio_pika.Exchange = None    
-    progress_exchange: aio_pika.Exchange = None
+    _logs_exchange: aio_pika.Exchange = None
+    _progress_exchange: aio_pika.Exchange = None
+
+    class Config:
+
+        arbitrary_types_allowed = True
 
     async def connect(self):
         url = eval_broker(self._config)
         await wait_till_rabbit_responsive(url)
 
-
         self._connection = await aio_pika.connect_robust(
-            url,
-            client_properties={"connection_name": "sidecar connection"},
+            url, client_properties={"connection_name": "sidecar connection"},
         )
 
         self._channel = await self._connection.channel()
-        self.logs_exchange = await self._channel.declare_exchange(
+        self._logs_exchange = await self._channel.declare_exchange(
             self._config.log_channel, aio_pika.ExchangeType.FANOUT, auto_delete=True
         )
-        self.progress_exchange = await self._channel.declare_exchange(
-            self._config.progress_channel, aio_pika.ExchangeType.FANOUT, auto_delete=True
+        self._progress_exchange = await self._channel.declare_exchange(
+            self._config.progress_channel,
+            aio_pika.ExchangeType.FANOUT,
+            auto_delete=True,
         )
 
-    async def post_message(self):
-        
+    async def close(self):
+        await self._channel.close()
+        await self._connection.close()
+
+    async def _post_message(self, exchange: aio_pika.Exchange, data: Dict[str, str]):
+        await exchange.publish(message=json.dumps(data))
+
+    async def post_log_message(
+        self,
+        user_id: str,
+        project_id: str,
+        node_id: str,
+        log_msg: Union[str, List[str]],
+    ):
+        await self._post_message(
+            self._logs_exchange,
+            data={
+                "Channel": "Log",
+                "Node": node_id,
+                "user_id": user_id,
+                "project_id": project_id,
+                "Messages": log_msg if isinstance(log_msg, list) else [log_msg],
+            },
+        )
+
+    async def post_progress_message(
+        self, user_id: str, project_id: str, node_id: str, progress_msg: str
+    ):
+        await self._post_message(
+            self._logs_exchange,
+            data={
+                "Channel": "Progress",
+                "Node": node_id,
+                "user_id": user_id,
+                "project_id": project_id,
+                "Messages": progress_msg,
+            },
+        )
 
 
 @tenacity.retry(**RabbitMQRetryPolicyUponInitialization().kwargs)
 async def wait_till_rabbit_responsive(url: str):
     connection = await aio_pika.connect(url)
     await connection.close()
-    return True
+    return True

services/sidecar/tests/conftest.py

@@ -15,7 +15,7 @@
 
 @pytest.fixture(scope="session")
 def project_slug_dir():
-    folder = current_dir.parent.parent
+    folder = current_dir.parent
     assert folder.exists()
     assert any(folder.glob("src/simcore_service_sidecar"))
     return folder