remove log_catch again

Author: bisgaard-itis

.github/workflows/ci-testing-deploy.yml

@@ -514,7 +514,7 @@ jobs:
   unit-test-storage:
     needs: changes
     if: ${{ needs.changes.outputs.storage == 'true' || github.event_name == 'push' }}
-    timeout-minutes: 18 # if this timeout gets too small, then split the tests
+    timeout-minutes: 25 # if this timeout gets too small, then split the tests
     name: "[unit] storage"
     runs-on: ${{ matrix.os }}
     strategy:

packages/pytest-simcore/src/pytest_simcore/docker_compose.py

@@ -2,10 +2,10 @@
 # pylint: disable=unused-argument
 # pylint: disable=unused-variable
 
-""" Fixtures to create docker-compose.yaml configuration files (as in Makefile)
+"""Fixtures to create docker-compose.yaml configuration files (as in Makefile)
 
-    - Basically runs `docker compose config
-    - Services in stack can be selected using 'core_services_selection', 'ops_services_selection' fixtures
+- Basically runs `docker compose config
+- Services in stack can be selected using 'core_services_selection', 'ops_services_selection' fixtures
 
 """
 
@@ -391,6 +391,10 @@ def _filter_services_and_dump(
         if "environment" in service:
             service["environment"] = _minio_fix(service["environment"])
 
+        if name == "postgres":
+            # NOTE: # -c fsync=off is not recommended for production as this disable writing to disk https://pythonspeed.com/articles/faster-db-tests/
+            service["command"] += ["-c", "fsync=off"]
+
     # updates current docker-compose (also versioned ... do not change by hand)
     with docker_compose_path.open("wt") as fh:
         yaml.dump(content, fh, default_flow_style=False)

packages/service-library/src/servicelib/rabbitmq/_client_rpc.py

@@ -32,15 +32,18 @@ async def create(
         cls, *, client_name: str, settings: RabbitSettings, **kwargs
     ) -> "RabbitMQRPCClient":
         client = cls(client_name=client_name, settings=settings, **kwargs)
-        await client._rpc_initialize()  # noqa: SLF001
+        await client._rpc_initialize()
         return client
 
     async def _rpc_initialize(self) -> None:
+        # NOTE: to show the connection name in the rabbitMQ UI see there
+        # https://www.bountysource.com/issues/89342433-setting-custom-connection-name-via-client_properties-doesn-t-work-when-connecting-using-an-amqp-url
+        #
+        connection_name = f"{get_rabbitmq_client_unique_name(self.client_name)}.rpc"
+        url = f"{self.settings.dsn}?name={connection_name}"
         self._connection = await aio_pika.connect_robust(
-            self.settings.dsn,
-            client_properties={
-                "connection_name": f"{get_rabbitmq_client_unique_name(self.client_name)}.rpc"
-            },
+            url,
+            client_properties={"connection_name": connection_name},
         )
         self._channel = await self._connection.channel()
 

packages/settings-library/src/settings_library/redis.py

@@ -42,6 +42,6 @@ def build_redis_dsn(self, db_index: RedisDatabase) -> str:
                 ),
                 host=self.REDIS_HOST,
                 port=self.REDIS_PORT,
-                path=f"/{db_index}",
+                path=f"{db_index}",
             )
         )

services/storage/src/simcore_service_storage/api/rpc/_async_jobs.py

@@ -47,19 +47,18 @@ async def _assert_job_exists(
 async def cancel(app: FastAPI, job_id: AsyncJobId, job_id_data: AsyncJobNameData):
     assert app  # nosec
     assert job_id_data  # nosec
-    with log_catch(logger=_logger):
-        try:
-            await _assert_job_exists(
-                job_id=job_id,
-                job_id_data=job_id_data,
-                celery_client=get_celery_client(app),
-            )
-            await get_celery_client(app).abort_task(
-                task_context=job_id_data.model_dump(),
-                task_uuid=job_id,
-            )
-        except CeleryError as exc:
-            raise JobSchedulerError(exc=f"{exc}") from exc
+    try:
+        await _assert_job_exists(
+            job_id=job_id,
+            job_id_data=job_id_data,
+            celery_client=get_celery_client(app),
+        )
+        await get_celery_client(app).abort_task(
+            task_context=job_id_data.model_dump(),
+            task_uuid=job_id,
+        )
+    except CeleryError as exc:
+        raise JobSchedulerError(exc=f"{exc}") from exc
 
 
 @router.expose(reraise_if_error_type=(JobSchedulerError, JobMissingError))
@@ -69,25 +68,24 @@ async def status(
     assert app  # nosec
     assert job_id_data  # nosec
 
-    with log_catch(logger=_logger):
-        try:
-            await _assert_job_exists(
-                job_id=job_id,
-                job_id_data=job_id_data,
-                celery_client=get_celery_client(app),
-            )
-            task_status = await get_celery_client(app).get_task_status(
-                task_context=job_id_data.model_dump(),
-                task_uuid=job_id,
-            )
-        except CeleryError as exc:
-            raise JobSchedulerError(exc=f"{exc}") from exc
-
-        return AsyncJobStatus(
+    try:
+        await _assert_job_exists(
             job_id=job_id,
-            progress=task_status.progress_report,
-            done=task_status.is_done,
+            job_id_data=job_id_data,
+            celery_client=get_celery_client(app),
+        )
+        task_status = await get_celery_client(app).get_task_status(
+            task_context=job_id_data.model_dump(),
+            task_uuid=job_id,
         )
+    except CeleryError as exc:
+        raise JobSchedulerError(exc=f"{exc}") from exc
+
+    return AsyncJobStatus(
+        job_id=job_id,
+        progress=task_status.progress_report,
+        done=task_status.is_done,
+    )
 
 
 @router.expose(
@@ -106,38 +104,35 @@ async def result(
     assert job_id  # nosec
     assert job_id_data  # nosec
 
-    with log_catch(logger=_logger):
-        try:
-            await _assert_job_exists(
-                job_id=job_id,
-                job_id_data=job_id_data,
-                celery_client=get_celery_client(app),
-            )
-            _status = await get_celery_client(app).get_task_status(
-                task_context=job_id_data.model_dump(),
-                task_uuid=job_id,
-            )
-            if not _status.is_done:
-                raise JobNotDoneError(job_id=job_id)
-            _result = await get_celery_client(app).get_task_result(
-                task_context=job_id_data.model_dump(),
-                task_uuid=job_id,
-            )
-        except CeleryError as exc:
-            raise JobSchedulerError(exc=f"{exc}") from exc
-
-        if _status.task_state == TaskState.ABORTED:
-            raise JobAbortedError(job_id=job_id)
-        if _status.task_state == TaskState.ERROR:
-            exc_type = ""
-            exc_msg = ""
-            with log_catch(logger=_logger, reraise=False):
-                task_error = TaskError.model_validate(_result)
-                exc_type = task_error.exc_type
-                exc_msg = task_error.exc_msg
-            raise JobError(job_id=job_id, exc_type=exc_type, exc_msg=exc_msg)
-
-        return AsyncJobResult(result=_result)
+    try:
+        await _assert_job_exists(
+            job_id=job_id, job_id_data=job_id_data, celery_client=get_celery_client(app)
+        )
+        _status = await get_celery_client(app).get_task_status(
+            task_context=job_id_data.model_dump(),
+            task_uuid=job_id,
+        )
+        if not _status.is_done:
+            raise JobNotDoneError(job_id=job_id)
+        _result = await get_celery_client(app).get_task_result(
+            task_context=job_id_data.model_dump(),
+            task_uuid=job_id,
+        )
+    except CeleryError as exc:
+        raise JobSchedulerError(exc=f"{exc}") from exc
+
+    if _status.task_state == TaskState.ABORTED:
+        raise JobAbortedError(job_id=job_id)
+    if _status.task_state == TaskState.ERROR:
+        exc_type = ""
+        exc_msg = ""
+        with log_catch(logger=_logger, reraise=False):
+            task_error = TaskError.model_validate(_result)
+            exc_type = task_error.exc_type
+            exc_msg = task_error.exc_msg
+        raise JobError(job_id=job_id, exc_type=exc_type, exc_msg=exc_msg)
+
+    return AsyncJobResult(result=_result)
 
 
 @router.expose(reraise_if_error_type=(JobSchedulerError,))
@@ -146,12 +141,11 @@ async def list_jobs(
 ) -> list[AsyncJobGet]:
     assert app  # nosec
     assert filter_  # nosec
-    with log_catch(logger=_logger):
-        try:
-            task_uuids = await get_celery_client(app).get_task_uuids(
-                task_context=job_id_data.model_dump(),
-            )
-        except CeleryError as exc:
-            raise JobSchedulerError(exc=f"{exc}") from exc
-
-        return [AsyncJobGet(job_id=task_uuid) for task_uuid in task_uuids]
+    try:
+        task_uuids = await get_celery_client(app).get_task_uuids(
+            task_context=job_id_data.model_dump(),
+        )
+    except CeleryError as exc:
+        raise JobSchedulerError(exc=f"{exc}") from exc
+
+    return [AsyncJobGet(job_id=task_uuid) for task_uuid in task_uuids]

services/storage/src/simcore_service_storage/api/rpc/_data_export.py

@@ -12,7 +12,6 @@
     DataExportTaskStartInput,
     InvalidFileIdentifierError,
 )
-from servicelib.logging_utils import log_catch
 from servicelib.rabbitmq import RPCRouter
 
 from ...datcore_dsm import DatCoreDataManager
@@ -41,31 +40,30 @@ async def start_data_export(
 ) -> AsyncJobGet:
     assert app  # nosec
 
-    with log_catch(_logger):
-        dsm = get_dsm_provider(app).get(data_export_start.location_id)
+    dsm = get_dsm_provider(app).get(data_export_start.location_id)
 
-        try:
-            for _id in data_export_start.file_and_folder_ids:
-                if isinstance(dsm, DatCoreDataManager):
-                    _ = await dsm.get_file(user_id=job_id_data.user_id, file_id=_id)
-                elif isinstance(dsm, SimcoreS3DataManager):
-                    await dsm.can_read_file(user_id=job_id_data.user_id, file_id=_id)
+    try:
+        for _id in data_export_start.file_and_folder_ids:
+            if isinstance(dsm, DatCoreDataManager):
+                _ = await dsm.get_file(user_id=job_id_data.user_id, file_id=_id)
+            elif isinstance(dsm, SimcoreS3DataManager):
+                await dsm.can_read_file(user_id=job_id_data.user_id, file_id=_id)
 
-        except (FileAccessRightError, DatcoreAdapterError) as err:
-            raise AccessRightError(
-                user_id=job_id_data.user_id,
-                file_id=_id,
-                location_id=data_export_start.location_id,
-            ) from err
+    except (FileAccessRightError, DatcoreAdapterError) as err:
+        raise AccessRightError(
+            user_id=job_id_data.user_id,
+            file_id=_id,
+            location_id=data_export_start.location_id,
+        ) from err
 
-        try:
-            task_uuid = await get_celery_client(app).send_task(
-                "export_data",
-                task_context=job_id_data.model_dump(),
-                files=data_export_start.file_and_folder_ids,  # ANE: adapt here your signature
-            )
-        except CeleryError as exc:
-            raise JobSchedulerError(exc=f"{exc}") from exc
-        return AsyncJobGet(
-            job_id=task_uuid,
+    try:
+        task_uuid = await get_celery_client(app).send_task(
+            "export_data",
+            task_context=job_id_data.model_dump(),
+            files=data_export_start.file_and_folder_ids,  # ANE: adapt here your signature
         )
+    except CeleryError as exc:
+        raise JobSchedulerError(exc=f"{exc}") from exc
+    return AsyncJobGet(
+        job_id=task_uuid,
+    )

services/storage/src/simcore_service_storage/api/rpc/_paths.py

@@ -7,7 +7,6 @@
     AsyncJobNameData,
 )
 from models_library.projects_nodes_io import LocationID
-from servicelib.logging_utils import log_catch
 from servicelib.rabbitmq import RPCRouter
 
 from ...modules.celery import get_celery_client
@@ -26,13 +25,12 @@ async def compute_path_size(
     path: Path,
 ) -> AsyncJobGet:
     assert app  # nosec
-    with log_catch(logger=_logger):
-        task_uuid = await get_celery_client(app).send_task(
-            remote_compute_path_size.__name__,
-            task_context=job_id_data.model_dump(),
-            user_id=job_id_data.user_id,
-            location_id=location_id,
-            path=path,
-        )
+    task_uuid = await get_celery_client(app).send_task(
+        remote_compute_path_size.__name__,
+        task_context=job_id_data.model_dump(),
+        user_id=job_id_data.user_id,
+        location_id=location_id,
+        path=path,
+    )
 
-        return AsyncJobGet(job_id=task_uuid)
+    return AsyncJobGet(job_id=task_uuid)