Merge branch 'master' into fix_duplicate_map_inputs

Author: wvangeit

.env-devel

@@ -377,7 +377,7 @@ TRACING_OPENTELEMETRY_COLLECTOR_BATCH_SIZE=2
 TRACING_OPENTELEMETRY_COLLECTOR_ENDPOINT=http://opentelemetry-collector
 TRACING_OPENTELEMETRY_COLLECTOR_EXPORTER_ENDPOINT=http://jaeger:4318
 TRACING_OPENTELEMETRY_COLLECTOR_PORT=4318
-TRACING_OPENTELEMETRY_COLLECTOR_SAMPLING_PERCENTAGE=100
+TRACING_OPENTELEMETRY_SAMPLING_PROBABILITY=1.0
 TRAEFIK_SIMCORE_ZONE=internal_simcore_stack
 TRASH_RETENTION_DAYS=7
 TWILIO_ACCOUNT_SID=DUMMY

packages/models-library/src/models_library/api_schemas_catalog/services.py

@@ -6,6 +6,7 @@
 from pydantic import ConfigDict, Field, HttpUrl, NonNegativeInt
 from pydantic.config import JsonDict
 
+from ..batch_operations import BatchGetEnvelope
 from ..boot_options import BootOptions
 from ..emails import LowerCaseEmailStr
 from ..groups import GroupID
@@ -420,6 +421,46 @@ class MyServiceGet(CatalogOutputSchema):
     my_access_rights: ServiceGroupAccessRightsV2
 
 
+class MyServicesRpcBatchGet(
+    CatalogOutputSchema,
+    BatchGetEnvelope[MyServiceGet, tuple[ServiceKey, ServiceVersion]],
+):
+    """Result for batch get user services operations"""
+
+    @staticmethod
+    def _update_json_schema_extra(schema: JsonDict) -> None:
+        missing: Any = [("simcore/services/comp/itis/sleeper", "100.2.3")]
+        schema.update(
+            {
+                "examples": [
+                    {
+                        "found_items": [
+                            {
+                                "key": missing[0][0],
+                                "release": {
+                                    "version": "2.2.1",
+                                    "version_display": "Winter Release",
+                                    "released": "2026-07-21T15:00:00",
+                                },
+                                "owner": 42,
+                                "my_access_rights": {
+                                    "execute": True,
+                                    "write": False,
+                                },
+                            }
+                        ],
+                        "missing_identifiers": missing,
+                    }
+                ]
+            }
+        )
+
+    model_config = ConfigDict(
+        extra="forbid",
+        json_schema_extra=_update_json_schema_extra,
+    )
+
+
 class ServiceListFilters(Filters):
     service_type: Annotated[
         ServiceType | None,

packages/models-library/src/models_library/api_schemas_webserver/projects_nodes.py

@@ -12,6 +12,7 @@
 from ..projects_nodes import InputID, InputsDict, PartialNode
 from ..projects_nodes_io import NodeID
 from ..services import ServiceKey, ServicePortKey, ServiceVersion
+from ..services_base import ServiceKeyVersion
 from ..services_enums import ServiceState
 from ..services_history import ServiceRelease
 from ..services_resources import ServiceResourcesDict
@@ -225,3 +226,9 @@ class NodeServiceGet(OutputSchema):
 class ProjectNodeServicesGet(OutputSchema):
     project_uuid: ProjectID
     services: list[NodeServiceGet]
+    missing: Annotated[
+        list[ServiceKeyVersion] | None,
+        Field(
+            description="List of services defined in the project but that were not found in the catalog"
+        ),
+    ] = None

packages/models-library/src/models_library/batch_operations.py

@@ -0,0 +1,62 @@
+from typing import Annotated, Generic, TypeVar
+
+from common_library.basic_types import DEFAULT_FACTORY
+from pydantic import BaseModel, BeforeValidator, Field, TypeAdapter
+
+ResourceT = TypeVar("ResourceT")
+IdentifierT = TypeVar("IdentifierT")
+SchemaT = TypeVar("SchemaT")
+
+
+def _deduplicate_preserving_order(identifiers: list[IdentifierT]) -> list[IdentifierT]:
+    """Remove duplicates while preserving order of first occurrence."""
+    return list(dict.fromkeys(identifiers))
+
+
+def create_batch_ids_validator(identifier_type: type[IdentifierT]) -> TypeAdapter:
+    """Create a TypeAdapter for validating batch identifiers.
+
+    This validator ensures:
+    - At least one identifier is provided (empty list is invalid for batch operations)
+    - Duplicates are removed while preserving order
+
+    Args:
+        identifier_type: The type of identifiers in the batch
+
+    Returns:
+        TypeAdapter configured for the specific identifier type
+    """
+    return TypeAdapter(
+        Annotated[
+            list[identifier_type],  # type: ignore[valid-type]
+            BeforeValidator(_deduplicate_preserving_order),
+            Field(
+                min_length=1,
+                description="List of identifiers to batch process. Empty list is not allowed for batch operations.",
+            ),
+        ]
+    )
+
+
+class BatchGetEnvelope(BaseModel, Generic[ResourceT, IdentifierT]):
+    """Generic envelope model for batch-get operations that can contain partial results.
+
+    This model represents the result of a batch operation where some items might be found
+    and others might be missing. It enforces that at least one item must be found,
+    as an empty batch operation is considered a client error.
+    """
+
+    found_items: Annotated[
+        list[ResourceT],
+        Field(
+            min_length=1,
+            description="List of successfully retrieved items. Must contain at least one item.",
+        ),
+    ]
+    missing_identifiers: Annotated[
+        list[IdentifierT],
+        Field(
+            default_factory=list,
+            description="List of identifiers for items that were not found",
+        ),
+    ] = DEFAULT_FACTORY

packages/models-library/tests/test_batch_operations.py

@@ -0,0 +1,108 @@
+import pytest
+from faker import Faker
+from models_library.api_schemas_webserver._base import (
+    OutputSchema as WebServerOutputSchema,
+)
+from models_library.api_schemas_webserver.projects import (
+    ProjectGet,
+)
+from models_library.batch_operations import BatchGetEnvelope, create_batch_ids_validator
+from models_library.generics import Envelope
+from models_library.projects import ProjectID
+from pydantic import TypeAdapter, ValidationError
+
+
+@pytest.mark.parametrize(
+    "identifier_type,input_ids,expected_output,should_raise",
+    [
+        # Valid cases - successful validation
+        pytest.param(
+            str, ["a", "b", "c"], ["a", "b", "c"], False, id="str_valid_no_duplicates"
+        ),
+        pytest.param(int, [1, 2, 3], [1, 2, 3], False, id="int_valid_no_duplicates"),
+        pytest.param(
+            tuple,
+            [("a", 1), ("b", 2)],
+            [("a", 1), ("b", 2)],
+            False,
+            id="tuple_valid_no_duplicates",
+        ),
+        # Deduplication cases - preserving order
+        pytest.param(
+            str, ["a", "b", "a", "c"], ["a", "b", "c"], False, id="str_with_duplicates"
+        ),
+        pytest.param(int, [1, 2, 1, 3, 2], [1, 2, 3], False, id="int_with_duplicates"),
+        pytest.param(
+            tuple,
+            [("a", 1), ("b", 2), ("a", 1)],
+            [("a", 1), ("b", 2)],
+            False,
+            id="tuple_with_duplicates",
+        ),
+        # Single item cases
+        pytest.param(str, ["single"], ["single"], False, id="str_single_item"),
+        pytest.param(int, [42], [42], False, id="int_single_item"),
+        # Edge case - all duplicates resolve to single item
+        pytest.param(
+            str, ["same", "same", "same"], ["same"], False, id="str_all_duplicates"
+        ),
+        # Error cases - empty list should raise ValidationError
+        pytest.param(str, [], None, True, id="str_empty_list_error"),
+        pytest.param(int, [], None, True, id="int_empty_list_error"),
+        pytest.param(tuple, [], None, True, id="tuple_empty_list_error"),
+    ],
+)
+def test_create_batch_ids_validator(
+    identifier_type, input_ids, expected_output, should_raise
+):
+    validator = create_batch_ids_validator(identifier_type)
+
+    if should_raise:
+        with pytest.raises(ValidationError) as exc_info:
+            validator.validate_python(input_ids)
+
+        # Verify the error is about minimum length
+        assert "at least 1" in str(exc_info.value).lower()
+    else:
+        result = validator.validate_python(input_ids)
+        assert result == expected_output
+        assert len(result) >= 1  # Ensure minimum length constraint
+        # Verify order preservation by checking first occurrence positions
+        if len(set(input_ids)) != len(input_ids):  # Had duplicates
+            original_first_positions = {
+                item: input_ids.index(item) for item in set(input_ids)
+            }
+            # Items should appear in the same relative order as their first occurrence
+            sorted_by_original = sorted(
+                result, key=lambda x: original_first_positions[x]
+            )
+            assert result == sorted_by_original
+
+
+def test_composing_schemas_for_batch_operations(faker: Faker):
+
+    # inner schema model
+    class WebServerProjectBatchGetSchema(
+        WebServerOutputSchema, BatchGetEnvelope[ProjectGet, ProjectID]
+    ): ...
+
+    some_projects = ProjectGet.model_json_schema()["examples"]
+
+    # response model
+    response_model = Envelope[WebServerProjectBatchGetSchema].model_validate(
+        {
+            # NOTE: how camelcase (from WebServerOutputSchema.model_config) applies here
+            "data": {
+                "foundItems": some_projects,
+                "missingIdentifiers": [ProjectID(faker.uuid4())],
+            }
+        }
+    )
+
+    assert response_model.data is not None
+
+    assert response_model.data.found_items == TypeAdapter(
+        list[ProjectGet]
+    ).validate_python(some_projects)
+
+    assert len(response_model.data.missing_identifiers) == 1

packages/service-library/src/servicelib/aiohttp/monitoring.py

@@ -20,6 +20,7 @@
 from ..prometheus_metrics import (
     PrometheusMetrics,
     get_prometheus_metrics,
+    record_asyncio_event_looop_metrics,
     record_request_metrics,
     record_response_metrics,
 )
@@ -33,15 +34,15 @@
 
 def get_collector_registry(app: web.Application) -> CollectorRegistry:
     metrics = app[PROMETHEUS_METRICS_APPKEY]
-    assert isinstance(metrics, PrometheusMetrics)  # nosec
     return metrics.registry
 
 
 async def metrics_handler(request: web.Request):
-    registry = get_collector_registry(request.app)
+    metrics = request.app[PROMETHEUS_METRICS_APPKEY]
+    await record_asyncio_event_looop_metrics(metrics)
 
     # NOTE: Cannot use ProcessPoolExecutor because registry is not pickable
-    result = await request.loop.run_in_executor(None, generate_latest, registry)
+    result = await request.loop.run_in_executor(None, generate_latest, metrics.registry)
     response = web.Response(body=result)
     response.content_type = CONTENT_TYPE_LATEST
     return response