Renamed comp_backend as computation

Minor

Author: Pedro Crespo

services/web/server/src/simcore_service_webserver/computation.py

@@ -11,7 +11,7 @@
 from aiohttp import web
 
 from .application_keys import APP_CONFIG_KEY
-from .comp_backend_subscribe import subscribe
+from .computation_subscribe import subscribe
 from .computation_config import CONFIG_SECTION_NAME, SERVICE_NAME
 
 log = logging.getLogger(__file__)

services/web/server/src/simcore_service_webserver/comp_backend_api.py renamed to services/web/server/src/simcore_service_webserver/computation_api.py

@@ -20,7 +20,7 @@
 
 from . import director_sdk
 from .application_keys import APP_CONFIG_KEY
-from .comp_backend_worker import celery
+from .computation_worker import celery
 
 # TODO: this should be coordinated with postgres options from config/server.yaml
 #from simcore_sdk.config.db import Config as DbConfig
@@ -34,7 +34,7 @@
 
 
 db_session = None
-comp_backend_routes = web.RouteTableDef()
+computation_routes = web.RouteTableDef()
 
 async def init_database(_app):
     #pylint: disable=W0603
@@ -88,7 +88,7 @@ async def _get_node_details(node_key:str, node_version:str)->dict:
     try:
         services_enveloped = await director_sdk.get_director().services_by_key_version_get(node_key, node_version)
         node_details = services_enveloped.data[0].to_dict()
-        return node_details        
+        return node_details
     except ApiException as err:
         log.exception("Error could not find service %s:%s", node_key, node_version)
         raise web_exceptions.HTTPNotFound(reason=str(err))
@@ -120,7 +120,7 @@ async def _build_adjacency_list(node_uuid:str, node_schema:dict, node_inputs:dic
                 if node_uuid not in dag_adjacency_list[input_node_uuid] and is_node_computational:
                     dag_adjacency_list[input_node_uuid].append(node_uuid)
     return dag_adjacency_list
-    
+
 async def _parse_pipeline(pipeline_data:dict): # pylint: disable=R0912
     dag_adjacency_list = dict()
     tasks = dict()
@@ -133,7 +133,7 @@ async def _parse_pipeline(pipeline_data:dict): # pylint: disable=R0912
         node_key = value["key"]
         node_version = value["version"]
 
-        # get the task data        
+        # get the task data
         node_inputs = None
         if "inputs" in value:
             node_inputs = value["inputs"]
@@ -167,20 +167,20 @@ async def _parse_pipeline(pipeline_data:dict): # pylint: disable=R0912
 
 async def _set_adjacency_in_pipeline_db(project_id, dag_adjacency_list):
     try:
-        pipeline = db_session.query(ComputationalPipeline).filter(ComputationalPipeline.project_id==project_id).one()            
+        pipeline = db_session.query(ComputationalPipeline).filter(ComputationalPipeline.project_id==project_id).one()
         log.debug("Pipeline object found")
         pipeline.state = 0
         pipeline.dag_adjacency_list = dag_adjacency_list
     except sqlalchemy.orm.exc.NoResultFound:
         # let's create one then
-        pipeline = ComputationalPipeline(project_id=project_id, dag_adjacency_list=dag_adjacency_list, state=0)    
+        pipeline = ComputationalPipeline(project_id=project_id, dag_adjacency_list=dag_adjacency_list, state=0)
         log.debug("Pipeline object created")
         db_session.add(pipeline)
     except sqlalchemy.orm.exc.MultipleResultsFound:
         log.exception("the computation pipeline %s is not unique", project_id)
         raise
 
-async def _set_tasks_in_tasks_db(project_id, tasks):    
+async def _set_tasks_in_tasks_db(project_id, tasks):
     tasks_db = db_session.query(ComputationalTask).filter(ComputationalTask.project_id==project_id).all()
     # delete tasks that were deleted from the db
     for task_db in tasks_db:
@@ -213,7 +213,7 @@ async def _set_tasks_in_tasks_db(project_id, tasks):
             db_session.add(comp_task)
 
 # pylint:disable=too-many-branches, too-many-statements
-@comp_backend_routes.post("/start_pipeline")
+@computation_routes.post("/start_pipeline")
 async def start_pipeline(request):
     #pylint:disable=broad-except
     # FIXME: this should be implemented generaly using async lazy initialization of db_session??
@@ -230,7 +230,7 @@ async def start_pipeline(request):
 
     log.debug("Client calls start_pipeline with project id: %s, pipeline data %s", project_id, pipeline_data)
     dag_adjacency_list, tasks = await _parse_pipeline(pipeline_data)
-    log.debug("Pipeline parsed:\nlist: %s\ntasks: %s", str(dag_adjacency_list), str(tasks))    
+    log.debug("Pipeline parsed:\nlist: %s\ntasks: %s", str(dag_adjacency_list), str(tasks))
     try:
         await _set_adjacency_in_pipeline_db(project_id, dag_adjacency_list)
         await _set_tasks_in_tasks_db(project_id, tasks)

services/web/server/src/simcore_service_webserver/comp_backend_subscribe.py renamed to services/web/server/src/simcore_service_webserver/computation_subscribe.py

@@ -3,12 +3,13 @@
 import logging
 
 import aio_pika
+from aiohttp import web
 
 from simcore_sdk.config.rabbit import eval_broker
 
 from .application_keys import APP_CONFIG_KEY
 from .computation_config import CONFIG_SECTION_NAME
-from .sockets import SIO
+from .sockets import sio
 
 log = logging.getLogger(__file__)
 
@@ -18,16 +19,16 @@ async def on_message(message: aio_pika.IncomingMessage):
         data = json.loads(message.body)
         log.debug(data)
         if data["Channel"] == "Log":
-            await SIO.emit("logger", data = json.dumps(data))
+            await sio.emit("logger", data = json.dumps(data))
         elif data["Channel"] == "Progress":
-            await SIO.emit("progress", data = json.dumps(data))
+            await sio.emit("progress", data = json.dumps(data))
 
-async def subscribe(_app=None):
+async def subscribe(app: web.Application):
     # TODO: catch and deal with missing connections:
     # e.g. CRITICAL:pika.adapters.base_connection:Could not get addresses to use: [Errno -2] Name or service not known (rabbit)
     # This exception is catch and pika persists ... WARNING:pika.connection:Could not connect, 5 attempts l
 
-    rb_config = _app[APP_CONFIG_KEY][CONFIG_SECTION_NAME]
+    rb_config = app[APP_CONFIG_KEY][CONFIG_SECTION_NAME]
     rabbit_broker = eval_broker(rb_config)
 
     # FIXME: This tmp resolves ``aio pika 169: IncompatibleProtocolError`` upon apio_pika.connect

services/web/server/src/simcore_service_webserver/comp_backend_worker.py renamed to services/web/server/src/simcore_service_webserver/computation_worker.py

@@ -15,7 +15,7 @@
 
 from .application_keys import (APP_CONFIG_KEY, APP_DB_ENGINE_KEY,
                                APP_DB_SESSION_KEY)
-from .comp_backend_api import init_database as _init_db
+from .computation_api import init_database as _init_db
 from .db_models import metadata
 from .db_config import CONFIG_SECTION_NAME
 

services/web/server/src/simcore_service_webserver/db.py

@@ -7,7 +7,7 @@
 logger = logging.getLogger(__name__)
 
 def setup(app: web.Application):
-    log.debug("Setting up %s ...", __name__)
+    logger.debug("Setting up %s ...", __name__)
 
     assert CONFIG_SECTION_NAME in app[APP_CONFIG_KEY]
 

services/web/server/src/simcore_service_webserver/director.py

@@ -10,7 +10,7 @@
 
 from servicelib import openapi
 
-from . import comp_backend_api, registry_api, rest_handlers
+from . import computation_api, registry_api, rest_handlers
 from .application_keys import APP_OPENAPI_SPECS_KEY
 from .login import routes as auth_routes
 from .rest_settings import get_base_path
@@ -45,7 +45,7 @@ def create(specs: openapi.Spec) -> List[web.RouteDef]:
     # FIXME: temp fix for running pipelines
     path, handle = '/services', registry_api.get_services
     routes.append(web.get(BASEPATH+path, handle))
-    path, handle = '/start_pipeline', comp_backend_api.start_pipeline
+    path, handle = '/start_pipeline', computation_api.start_pipeline
     routes.append(web.post(BASEPATH+path, handle))
 
     return routes