Merge branch 'master' into pr/giancarloromeo/8141

Author: pcrespov

.env-devel

@@ -56,6 +56,7 @@ CLUSTERS_KEEPER_COMPUTATIONAL_BACKEND_DEFAULT_CLUSTER_AUTH='{"type":"tls","tls_c
 CLUSTERS_KEEPER_COMPUTATIONAL_BACKEND_DOCKER_IMAGE_TAG=master-github-latest
 CLUSTERS_KEEPER_DASK_NPROCS=1
 CLUSTERS_KEEPER_DASK_NTHREADS=0
+CLUSTERS_KEEPER_DASK_NTHREADS_MULTIPLIER=1
 CLUSTERS_KEEPER_DASK_WORKER_SATURATION=inf
 CLUSTERS_KEEPER_EC2_ACCESS=null
 CLUSTERS_KEEPER_SSM_ACCESS=null

.github/instructions/python.instructions.md

@@ -35,6 +35,8 @@ Ensure compatibility with the following library versions:
 
 * Use `f-string` formatting for all string interpolation except for logging message strings.
 * Use **relative imports** within the same package/module.
+  - For imports within the same repository/project, always use relative imports (e.g., `from ..constants import APP_SETTINGS_KEY` instead of `from simcore_service_webserver.constants import APP_SETTINGS_KEY`)
+  - Use absolute imports only for external libraries and packages
 * Place **all imports at the top** of the file.
 * Document functions when the code is not self-explanatory or if asked explicitly.
 
@@ -44,6 +46,18 @@ Ensure compatibility with the following library versions:
 * Prefer `json_dumps` / `json_loads` from `common_library.json_serialization` instead of the built-in `json.dumps` / `json.loads`.
 * When using Pydantic models, prefer methods like `model.model_dump_json()` for serialization.
 
-### 7. **Running tests**
+### 7. **aiohttp Framework**
+
+* **Application Keys**: Always use `web.AppKey` for type-safe application storage instead of string keys
+  - Define keys with specific types: `APP_MY_KEY: Final = web.AppKey("APP_MY_KEY", MySpecificType)`
+  - Use precise types instead of generic `object` when the actual type is known
+  - Example: `APP_SETTINGS_KEY: Final = web.AppKey("APP_SETTINGS_KEY", ApplicationSettings)`
+  - Store and retrieve: `app[APP_MY_KEY] = value` and `data = app[APP_MY_KEY]`
+* **Request Keys**: Use `web.AppKey` for request storage as well for consistency and type safety
+* **Middleware**: Follow the repository's middleware patterns for cross-cutting concerns
+* **Error Handling**: Use the established exception handling decorators and patterns
+* **Route Definitions**: Use `web.RouteTableDef()` and organize routes logically within modules
+
+### 8. **Running tests**
 * Use `--keep-docker-up` flag when testing to keep docker containers up between sessions.
 * Always activate the python virtual environment before running pytest.

.github/prompts/refactor-aiohttp-appkey.prompt.md

@@ -0,0 +1,28 @@
+---
+mode: edit
+description: Converts string-based aiohttp app key constants to type-safe web.AppKey
+model: GPT-4.1
+---
+
+Convert all string-based app key constants to use type-safe web.AppKey.
+
+- Replace patterns like:
+  ```python
+  CONSTNAME_APPKEY: Final[str] = f"{__name__}.my_key"
+  ```
+  with:
+  ```python
+  from aiohttp import web
+  CONSTNAME_APPKEY: Final = web.AppKey("CONSTNAME", ValueType)
+  ```
+  (Replace ValueType with the actual type stored under this key.)
+
+- Update all usages:
+  - `app[CONSTNAME_APPKEY] = value`
+  - `data = app[CONSTNAME_APPKEY]` or `data = request.app[CONSTNAME_APPKEY]`
+
+- Key constant MUST be UPPERCASE
+- Key name MUST be suffixed `_APPKEY`
+- Remove any f"{__name__}..." patterns; use a simple string identifier in web.AppKey.
+- Ensure all keys are type-safe and self-documenting.
+- IF you change the original name, you MUST change all the references

.github/prompts/pydantic-annotated-fields.prompt.md renamed to .github/prompts/refactor-pydantic-annotated-fields.prompt.md

@@ -8,23 +8,23 @@ on:
 jobs:
   cleanup:
     runs-on: ubuntu-latest
+    permissions:
+      actions: write
     steps:
       - name: Cleanup
         run: |
-          gh extension install actions/gh-actions-cache
-
-          echo "Fetching list of cache key"
-          cacheKeysForPR=$(gh actions-cache list -R $REPO -B $BRANCH -L 100 | cut -f 1 )
+          echo "Fetching list of cache keys"
+          cacheKeysForPR=$(gh cache list --ref $BRANCH --limit 100 --json id --jq '.[].id')
 
           ## Setting this to not fail the workflow while deleting cache keys.
           set +e
           echo "Deleting caches..."
           for cacheKey in $cacheKeysForPR
           do
-              gh actions-cache delete $cacheKey -R $REPO -B $BRANCH --confirm
+              gh cache delete $cacheKey
           done
           echo "Done"
         env:
-          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          REPO: ${{ github.repository }}
+          GH_TOKEN: ${{ github.token }}
+          GH_REPO: ${{ github.repository }}
           BRANCH: refs/pull/${{ github.event.pull_request.number }}/merge

.github/workflows/cleanup-caches-by-branches.yml

@@ -27,6 +27,7 @@ tests: ## runs unit tests
 		--durations=10 \
 		--exitfirst \
 		--failed-first \
+		--keep-docker-up \
 		--pdb \
 		-vv \
 		$(CURDIR)/tests
@@ -41,6 +42,7 @@ tests-ci: ## runs unit tests
 		--cov-report=term-missing \
 		--cov-report=xml \
 		--junitxml=junit.xml -o junit_family=legacy \
+		--keep-docker-up \
 		--cov=celery_library \
 		--durations=10 \
 		--log-date-format="%Y-%m-%d %H:%M:%S" \

packages/celery-library/Makefile

@@ -6,12 +6,12 @@
 from models_library.progress_bar import ProgressReport
 from pydantic import ValidationError
 from servicelib.celery.models import (
+    WILDCARD,
+    ExecutionMetadata,
+    OwnerMetadata,
     Task,
-    TaskFilter,
     TaskID,
     TaskInfoStore,
-    TaskMetadata,
-    Wildcard,
 )
 from servicelib.redis import RedisClientSDK, handle_redis_returns_union_types
 
@@ -35,23 +35,23 @@ def __init__(self, redis_client_sdk: RedisClientSDK) -> None:
     async def create_task(
         self,
         task_id: TaskID,
-        task_metadata: TaskMetadata,
+        execution_metadata: ExecutionMetadata,
         expiry: timedelta,
     ) -> None:
         task_key = _build_key(task_id)
         await handle_redis_returns_union_types(
             self._redis_client_sdk.redis.hset(
                 name=task_key,
                 key=_CELERY_TASK_METADATA_KEY,
-                value=task_metadata.model_dump_json(),
+                value=execution_metadata.model_dump_json(),
             )
         )
         await self._redis_client_sdk.redis.expire(
             task_key,
             expiry,
         )
 
-    async def get_task_metadata(self, task_id: TaskID) -> TaskMetadata | None:
+    async def get_task_metadata(self, task_id: TaskID) -> ExecutionMetadata | None:
         raw_result = await handle_redis_returns_union_types(
             self._redis_client_sdk.redis.hget(
                 _build_key(task_id), _CELERY_TASK_METADATA_KEY
@@ -61,7 +61,7 @@ async def get_task_metadata(self, task_id: TaskID) -> TaskMetadata | None:
             return None
 
         try:
-            return TaskMetadata.model_validate_json(raw_result)
+            return ExecutionMetadata.model_validate_json(raw_result)
         except ValidationError as exc:
             _logger.debug(
                 "Failed to deserialize task metadata for task %s: %s", task_id, f"{exc}"
@@ -85,9 +85,9 @@ async def get_task_progress(self, task_id: TaskID) -> ProgressReport | None:
             )
             return None
 
-    async def list_tasks(self, task_filter: TaskFilter) -> list[Task]:
-        search_key = _CELERY_TASK_INFO_PREFIX + task_filter.create_task_id(
-            task_uuid=Wildcard()
+    async def list_tasks(self, owner_metadata: OwnerMetadata) -> list[Task]:
+        search_key = _CELERY_TASK_INFO_PREFIX + owner_metadata.model_dump_task_id(
+            task_uuid=WILDCARD
         )
 
         keys: list[str] = []
@@ -112,11 +112,11 @@ async def list_tasks(self, task_filter: TaskFilter) -> list[Task]:
                 continue
 
             with contextlib.suppress(ValidationError):
-                task_metadata = TaskMetadata.model_validate_json(raw_metadata)
+                execution_metadata = ExecutionMetadata.model_validate_json(raw_metadata)
                 tasks.append(
                     Task(
-                        uuid=TaskFilter.get_task_uuid(key),
-                        metadata=task_metadata,
+                        uuid=OwnerMetadata.get_task_uuid(key),
+                        metadata=execution_metadata,
                     )
                 )
 

packages/celery-library/src/celery_library/backends/redis.py

@@ -4,7 +4,6 @@
 
 from celery.exceptions import CeleryError  # type: ignore[import-untyped]
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
-    AsyncJobFilter,
     AsyncJobGet,
     AsyncJobId,
     AsyncJobResult,
@@ -17,7 +16,7 @@
     JobNotDoneError,
     JobSchedulerError,
 )
-from servicelib.celery.models import TaskFilter, TaskState
+from servicelib.celery.models import OwnerMetadata, TaskState
 from servicelib.celery.task_manager import TaskManager
 from servicelib.logging_utils import log_catch
 from servicelib.rabbitmq import RPCRouter
@@ -34,14 +33,13 @@
 
 @router.expose(reraise_if_error_type=(JobSchedulerError, JobMissingError))
 async def cancel(
-    task_manager: TaskManager, job_id: AsyncJobId, job_filter: AsyncJobFilter
+    task_manager: TaskManager, job_id: AsyncJobId, owner_metadata: OwnerMetadata
 ):
     assert task_manager  # nosec
-    assert job_filter  # nosec
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    assert owner_metadata  # nosec
     try:
         await task_manager.cancel_task(
-            task_filter=task_filter,
+            owner_metadata=owner_metadata,
             task_uuid=job_id,
         )
     except TaskNotFoundError as exc:
@@ -52,15 +50,14 @@ async def cancel(
 
 @router.expose(reraise_if_error_type=(JobSchedulerError, JobMissingError))
 async def status(
-    task_manager: TaskManager, job_id: AsyncJobId, job_filter: AsyncJobFilter
+    task_manager: TaskManager, job_id: AsyncJobId, owner_metadata: OwnerMetadata
 ) -> AsyncJobStatus:
     assert task_manager  # nosec
-    assert job_filter  # nosec
+    assert owner_metadata  # nosec
 
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
     try:
         task_status = await task_manager.get_task_status(
-            task_filter=task_filter,
+            owner_metadata=owner_metadata,
             task_uuid=job_id,
         )
     except TaskNotFoundError as exc:
@@ -85,23 +82,21 @@ async def status(
     )
 )
 async def result(
-    task_manager: TaskManager, job_id: AsyncJobId, job_filter: AsyncJobFilter
+    task_manager: TaskManager, job_id: AsyncJobId, owner_metadata: OwnerMetadata
 ) -> AsyncJobResult:
     assert task_manager  # nosec
     assert job_id  # nosec
-    assert job_filter  # nosec
-
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    assert owner_metadata  # nosec
 
     try:
         _status = await task_manager.get_task_status(
-            task_filter=task_filter,
+            owner_metadata=owner_metadata,
             task_uuid=job_id,
         )
         if not _status.is_done:
             raise JobNotDoneError(job_id=job_id)
         _result = await task_manager.get_task_result(
-            task_filter=task_filter,
+            owner_metadata=owner_metadata,
             task_uuid=job_id,
         )
     except TaskNotFoundError as exc:
@@ -134,13 +129,12 @@ async def result(
 
 @router.expose(reraise_if_error_type=(JobSchedulerError,))
 async def list_jobs(
-    task_manager: TaskManager, job_filter: AsyncJobFilter
+    task_manager: TaskManager, owner_metadata: OwnerMetadata
 ) -> list[AsyncJobGet]:
     assert task_manager  # nosec
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
     try:
         tasks = await task_manager.list_tasks(
-            task_filter=task_filter,
+            owner_metadata=owner_metadata,
         )
     except CeleryError as exc:
         raise JobSchedulerError(exc=f"{exc}") from exc