Merge branch 'master' into add-conversations

Author: matusdrobuliak66

packages/dask-task-models-library/src/dask_task_models_library/container_tasks/utils.py

@@ -0,0 +1,44 @@
+from typing import Final
+from uuid import uuid4
+
+from models_library.projects import ProjectID
+from models_library.projects_nodes_io import NodeID
+from models_library.services_types import ServiceKey, ServiceVersion
+from models_library.users import UserID
+from pydantic import TypeAdapter
+
+from ..models import DaskJobID
+
+
+def generate_dask_job_id(
+    service_key: ServiceKey,
+    service_version: ServiceVersion,
+    user_id: UserID,
+    project_id: ProjectID,
+    node_id: NodeID,
+) -> DaskJobID:
+    """creates a dask job id:
+    The job ID shall contain the user_id, project_id, node_id
+    Also, it must be unique
+    and it is shown in the Dask scheduler dashboard website
+    """
+    return DaskJobID(
+        f"{service_key}:{service_version}:userid_{user_id}:projectid_{project_id}:nodeid_{node_id}:uuid_{uuid4()}"
+    )
+
+
+_JOB_ID_PARTS: Final[int] = 6
+
+
+def parse_dask_job_id(
+    job_id: str,
+) -> tuple[ServiceKey, ServiceVersion, UserID, ProjectID, NodeID]:
+    parts = job_id.split(":")
+    assert len(parts) == _JOB_ID_PARTS  # nosec
+    return (
+        parts[0],
+        parts[1],
+        TypeAdapter(UserID).validate_python(parts[2][len("userid_") :]),
+        ProjectID(parts[3][len("projectid_") :]),
+        NodeID(parts[4][len("nodeid_") :]),
+    )

packages/dask-task-models-library/src/dask_task_models_library/models.py

@@ -0,0 +1,4 @@
+from typing import TypeAlias
+
+DaskJobID: TypeAlias = str
+DaskResources: TypeAlias = dict[str, int | float]

packages/dask-task-models-library/tests/container_tasks/test_utils.py

@@ -0,0 +1,68 @@
+# pylint: disable=too-many-positional-arguments
+# pylint:disable=redefined-outer-name
+# pylint:disable=too-many-arguments
+# pylint:disable=unused-argument
+# pylint:disable=unused-variable
+
+import pytest
+from dask_task_models_library.container_tasks.utils import (
+    generate_dask_job_id,
+    parse_dask_job_id,
+)
+from faker import Faker
+from models_library.projects import ProjectID
+from models_library.projects_nodes_io import NodeID
+from models_library.services_types import ServiceKey, ServiceVersion
+from models_library.users import UserID
+from pydantic import TypeAdapter
+
+
+@pytest.fixture(
+    params=["simcore/service/comp/some/fake/service/key", "dockerhub-style/service_key"]
+)
+def service_key(request) -> ServiceKey:
+    return request.param
+
+
+@pytest.fixture()
+def service_version() -> str:
+    return "1234.32432.2344"
+
+
+@pytest.fixture
+def user_id(faker: Faker) -> UserID:
+    return TypeAdapter(UserID).validate_python(faker.pyint(min_value=1))
+
+
+@pytest.fixture
+def project_id(faker: Faker) -> ProjectID:
+    return ProjectID(faker.uuid4())
+
+
+@pytest.fixture
+def node_id(faker: Faker) -> NodeID:
+    return NodeID(faker.uuid4())
+
+
+def test_dask_job_id_serialization(
+    service_key: ServiceKey,
+    service_version: ServiceVersion,
+    user_id: UserID,
+    project_id: ProjectID,
+    node_id: NodeID,
+):
+    dask_job_id = generate_dask_job_id(
+        service_key, service_version, user_id, project_id, node_id
+    )
+    (
+        parsed_service_key,
+        parsed_service_version,
+        parsed_user_id,
+        parsed_project_id,
+        parsed_node_id,
+    ) = parse_dask_job_id(dask_job_id)
+    assert service_key == parsed_service_key
+    assert service_version == parsed_service_version
+    assert user_id == parsed_user_id
+    assert project_id == parsed_project_id
+    assert node_id == parsed_node_id

packages/pytest-simcore/src/pytest_simcore/helpers/playwright.py

@@ -401,6 +401,7 @@ class SocketIONodeProgressCompleteWaiter:
     )
     _last_progress_time: datetime = field(default_factory=lambda: datetime.now(tz=UTC))
     _received_messages: list[SocketIOEvent] = field(default_factory=list)
+    _result: bool = False
 
     def __call__(self, message: str) -> bool:
         # socket.io encodes messages like so
@@ -422,6 +423,7 @@ def __call__(self, message: str) -> bool:
                     self.node_id,
                     decoded_message.obj["service_state"],
                 )
+                self._result = False
                 return True
             if decoded_message.name == _OSparcMessages.NODE_PROGRESS.value:
                 node_progress_event = retrieve_node_progress_from_decoded_message(
@@ -449,13 +451,11 @@ def __call__(self, message: str) -> bool:
                             f"{json.dumps({k: round(v, 2) for k, v in self._current_progress.items()})}",
                         )
 
-                progress_completed = self.got_expected_node_progress_types() and all(
-                    round(progress, 1) == 1.0
-                    for progress in self._current_progress.values()
-                )
-                if progress_completed:
+                done = self._completed_successfully()
+                if done:
+                    self._result = True  # NOTE: might have failed but it is not sure. so we set the result to True
                     self.logger.info("✅ Service start completed successfully!! ✅")
-                return progress_completed
+                return done
 
         time_since_last_progress = datetime.now(UTC) - self._last_progress_time
         if time_since_last_progress > self.max_idle_timeout:
@@ -465,19 +465,22 @@ def __call__(self, message: str) -> bool:
                 time_since_last_progress,
                 self.node_id,
             )
+            self._result = True
             return True
 
         return False
 
-    def got_expected_node_progress_types(self) -> bool:
+    def _completed_successfully(self) -> bool:
         return all(
             progress_type in self._current_progress
             for progress_type in NodeProgressType.required_types_for_started_service()
+        ) and all(
+            round(progress, 1) == 1.0 for progress in self._current_progress.values()
         )
 
     @property
-    def number_received_messages(self) -> int:
-        return len(self._received_messages)
+    def success(self) -> bool:
+        return self._result
 
 
 def wait_for_service_endpoint_responding(
@@ -597,7 +600,9 @@ def expected_service_running(
                 msg=f"Waiting for node to run. Timeout: {timeout}",
             )
         )
+
         if is_service_legacy:
+            waiter = None
             ctx.logger.info(
                 "⚠️ Legacy service detected. We are skipping websocket messages in this case! ⚠️"
             )
@@ -612,7 +617,11 @@ def expected_service_running(
         service_running = ServiceRunning(iframe_locator=None)
         if press_start_button:
             _trigger_service_start(page, node_id)
+        yield service_running
+
     elapsed_time = arrow.utcnow() - started
+    if waiter and not waiter.success:
+        pytest.fail("❌ Service failed starting!  ❌")
 
     wait_for_service_endpoint_responding(
         node_id,
@@ -652,6 +661,7 @@ def wait_for_service_running(
             )
         )
         if is_service_legacy:
+            waiter = None
             ctx.logger.info(
                 "⚠️ Legacy service detected. We are skipping websocket messages in this case! ⚠️"
             )
@@ -666,6 +676,9 @@ def wait_for_service_running(
         if press_start_button:
             _trigger_service_start(page, node_id)
     elapsed_time = arrow.utcnow() - started
+
+    if waiter and not waiter.success:
+        pytest.fail("❌ Service failed starting!  ❌")
     wait_for_service_endpoint_responding(
         node_id,
         api_request_context=page.request,

services/api-server/requirements/_base.in

@@ -17,7 +17,6 @@
 --requirement ../../../packages/service-library/requirements/_fastapi.in
 
 aiofiles
-aiopg[sa]
 cryptography
 fastapi-pagination
 fastapi[all]

services/api-server/requirements/_base.txt

@@ -55,8 +55,6 @@ aiohttp==3.11.10
     #   -c requirements/../../../requirements/constraints.txt
     #   -r requirements/../../../packages/simcore-sdk/requirements/_base.in
     #   aiodocker
-aiopg==1.4.0
-    # via -r requirements/_base.in
 aiormq==6.8.1
     # via aio-pika
 aiosignal==1.3.1
@@ -84,8 +82,6 @@ arrow==1.3.0
     #   -r requirements/../../../packages/simcore-sdk/requirements/../../../packages/service-library/requirements/_base.in
 asgiref==3.8.1
     # via opentelemetry-instrumentation-asgi
-async-timeout==4.0.3
-    # via aiopg
 asyncpg==0.30.0
     # via sqlalchemy
 attrs==24.2.0
@@ -513,9 +509,7 @@ psutil==6.1.0
     #   -r requirements/../../../packages/service-library/requirements/_base.in
     #   -r requirements/../../../packages/simcore-sdk/requirements/../../../packages/service-library/requirements/_base.in
 psycopg2-binary==2.9.10
-    # via
-    #   aiopg
-    #   sqlalchemy
+    # via sqlalchemy
 pycparser==2.22
     # via cffi
 pycryptodome==3.21.0
@@ -785,7 +779,6 @@ sqlalchemy==1.4.54
     #   -r requirements/../../../packages/postgres-database/requirements/_base.in
     #   -r requirements/../../../packages/simcore-sdk/requirements/../../../packages/postgres-database/requirements/_base.in
     #   -r requirements/../../../packages/simcore-sdk/requirements/_base.in
-    #   aiopg
     #   alembic
 starlette==0.41.3
     # via

services/api-server/src/simcore_service_api_server/api/dependencies/authentication.py

@@ -6,8 +6,8 @@
 from models_library.products import ProductName
 from pydantic import BaseModel, PositiveInt
 
-from ...db.repositories.api_keys import ApiKeysRepository, UserAndProductTuple
-from ...db.repositories.users import UsersRepository
+from ...repository.api_keys import ApiKeysRepository, UserAndProductTuple
+from ...repository.users import UsersRepository
 from .database import get_repository
 
 # SEE https://swagger.io/docs/specification/authentication/basic-authentication/
@@ -22,9 +22,9 @@ class Identity(BaseModel):
 
 def _create_exception() -> HTTPException:
     _unauthorized_headers = {
-        "WWW-Authenticate": f'Basic realm="{basic_scheme.realm}"'
-        if basic_scheme.realm
-        else "Basic"
+        "WWW-Authenticate": (
+            f'Basic realm="{basic_scheme.realm}"' if basic_scheme.realm else "Basic"
+        )
     }
     return HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
@@ -46,7 +46,7 @@ async def get_current_identity(
     if user_and_product is None:
         exc = _create_exception()
         raise exc
-    email = await users_repo.get_active_user_email(user_and_product.user_id)
+    email = await users_repo.get_active_user_email(user_id=user_and_product.user_id)
     if not email:
         exc = _create_exception()
         raise exc

services/api-server/src/simcore_service_api_server/api/dependencies/database.py

@@ -1,53 +1,38 @@
 import logging
 from collections.abc import AsyncGenerator, Callable
-from typing import Annotated, cast
+from typing import Annotated
 
-from aiopg.sa import Engine
 from fastapi import Depends
 from fastapi.requests import Request
+from simcore_postgres_database.utils_aiosqlalchemy import (
+    get_pg_engine_stateinfo,
+)
 from sqlalchemy.ext.asyncio import AsyncEngine
 
-from ...db.events import get_asyncpg_engine
-from ...db.repositories import BaseRepository
-
-logger = logging.getLogger(__name__)
-
+from ...clients.postgres import get_engine
+from ...repository import BaseRepository
 
-def get_db_engine(request: Request) -> Engine:
-    return cast(Engine, request.app.state.engine)
+_logger = logging.getLogger(__name__)
 
 
 def get_db_asyncpg_engine(request: Request) -> AsyncEngine:
-    return get_asyncpg_engine(request.app)
+    return get_engine(request.app)
 
 
 def get_repository(repo_type: type[BaseRepository]) -> Callable:
     async def _get_repo(
-        engine: Annotated[Engine, Depends(get_db_engine)],
+        engine: Annotated[AsyncEngine, Depends(get_db_asyncpg_engine)],
     ) -> AsyncGenerator[BaseRepository, None]:
         # NOTE: 2 different ideas were tried here with not so good
         # 1st one was acquiring a connection per repository which lead to the following issue https://github.com/ITISFoundation/osparc-simcore/pull/1966
         # 2nd one was acquiring a connection per request which works but blocks the director-v2 responsiveness once
         # the max amount of connections is reached
         # now the current solution is to acquire connection when needed.
+        _logger.debug(
+            "Setting up a repository. Current state of connections: %s",
+            await get_pg_engine_stateinfo(engine),
+        )
 
-        available_engines = engine.maxsize - (engine.size - engine.freesize)
-        if available_engines <= 1:
-            logger.warning(
-                "Low pg connections available in pool: pool size=%d, acquired=%d, free=%d, reserved=[%d, %d]",
-                engine.size,
-                engine.size - engine.freesize,
-                engine.freesize,
-                engine.minsize,
-                engine.maxsize,
-            )
         yield repo_type(db_engine=engine)
 
     return _get_repo
-
-
-__all__: tuple[str, ...] = (
-    "Engine",
-    "get_db_engine",
-    "get_repository",
-)