jobs service and repo layers

Author: pcrespov

services/web/server/src/simcore_service_webserver/projects/_controller/projects_rpc.py

@@ -7,6 +7,7 @@
 from servicelib.rabbitmq import RPCRouter
 
 from ...rabbitmq import get_rabbitmq_rpc_server
+from .. import _jobs_service
 
 router = RPCRouter()
 
@@ -21,11 +22,14 @@ async def mark_project_as_job(
     project_uuid: ProjectID,
     job_parent_resource_name: str,
 ) -> None:
-    msg = (
-        f"You have reached {__name__} but this feature is still not implemented. "
-        f"Inputs: {app=}, {product_name=}, {user_id=}, {project_uuid=}, {job_parent_resource_name=}"
+
+    await _jobs_service.set_project_as_job(
+        app,
+        product_name=product_name,
+        user_id=user_id,
+        project_uuid=project_uuid,
+        job_parent_resource_name=job_parent_resource_name,
     )
-    raise NotImplementedError(msg)
 
 
 async def register_rpc_routes_on_startup(app: web.Application):

services/web/server/src/simcore_service_webserver/projects/_jobs_repository.py

@@ -0,0 +1,39 @@
+import logging
+
+from models_library.projects import ProjectID
+from simcore_postgres_database.models.projects_to_jobs import projects_to_jobs
+from simcore_postgres_database.utils_repos import transaction_context
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+from ..db.base_repository import BaseRepository
+
+_logger = logging.getLogger(__name__)
+
+
+class ProjectJobsRepository(BaseRepository):
+
+    async def set_project_as_job(
+        self,
+        connection: AsyncConnection | None = None,
+        *,
+        project_uuid: ProjectID,
+        job_parent_resource_name: str,
+    ) -> int:
+        async with transaction_context(self.engine, connection) as conn:
+            stmt = (
+                pg_insert(projects_to_jobs)
+                .values(
+                    project_uuid=project_uuid,
+                    job_parent_resource_name=job_parent_resource_name,
+                )
+                .on_conflict_do_update(
+                    index_elements=["project_uuid", "job_parent_resource_name"],
+                    set_={"job_parent_resource_name": job_parent_resource_name},
+                )
+                .returning(projects_to_jobs.c.id)
+            )
+
+            result = await conn.execute(stmt)
+            row = result.one()
+            return row.id

services/web/server/src/simcore_service_webserver/projects/_jobs_service.py

@@ -0,0 +1,48 @@
+import logging
+from typing import Annotated
+
+from aiohttp import web
+from models_library.products import ProductName
+from models_library.projects import ProjectID
+from models_library.users import UserID
+from pydantic import AfterValidator, validate_call
+
+from ._access_rights_service import check_user_project_permission
+from ._jobs_repository import ProjectJobsRepository
+
+_logger = logging.getLogger(__name__)
+
+
+def _validate_job_parent_resource_name(value: str) -> str:
+    if value and not value.startswith("/") and not value.endswith("/") and "/" in value:
+        return value
+    msg = "Invalid format: must contain '/' but cannot start or end with '/'"
+    raise ValueError(msg)
+
+
+@validate_call(config={"arbitrary_types_allowed": True})
+async def set_project_as_job(
+    app: web.Application,
+    *,
+    product_name: ProductName,
+    user_id: UserID,
+    project_uuid: ProjectID,
+    job_parent_resource_name: Annotated[
+        str, AfterValidator(_validate_job_parent_resource_name)
+    ],
+) -> None:
+
+    await check_user_project_permission(
+        app,
+        project_id=project_uuid,
+        user_id=user_id,
+        product_name=product_name,
+        permission="write",
+    )
+
+    repo = ProjectJobsRepository.create_from_app(app)
+
+    projects_to_jobs_id = await repo.set_project_as_job(
+        project_uuid=project_uuid, job_parent_resource_name=job_parent_resource_name
+    )
+    assert projects_to_jobs_id  # nosec