19145569596

Committed 06 Nov 2025 06:17PM UTC coverage: 86.819% (-0.03%) from 86.85%

Build # 19145569596

Build Type

Pull #1101

github

Committed by

web-flow

Commit Message

Merge 252d68849 into fabb4724c

Pull Request Pull Request #1101: feat: indicate supported platforms when checking a session image

Run Details

241 of 314 new or added lines in 12 files covered. (76.75%)

8 existing lines in 4 files now uncovered.

23054 of 26554 relevant lines covered (86.82%)

1.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.65

/components/renku_data_services/notebooks/blueprints.py

"""Notebooks service API."""

from dataclasses import dataclass

from sanic import Request, empty, exceptions, json
from sanic.response import HTTPResponse, JSONResponse
from sanic_ext import validate

from renku_data_services import base_models
from renku_data_services.app_config import logging
from renku_data_services.base_api.auth import authenticate, authenticate_2
from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint
from renku_data_services.base_models import AnonymousAPIUser, APIUser, AuthenticatedAPIUser, Authenticator
from renku_data_services.base_models.metrics import MetricsService
from renku_data_services.connected_services.models import ConnectionStatus
from renku_data_services.crc.db import ClusterRepository, ResourcePoolRepository
from renku_data_services.data_connectors.db import (
    DataConnectorRepository,
    DataConnectorSecretRepository,
)
from renku_data_services.errors import errors
from renku_data_services.notebooks import apispec, core
from renku_data_services.notebooks.api.classes.image import Image
from renku_data_services.notebooks.api.schemas.config_server_options import ServerOptionsEndpointResponse
from renku_data_services.notebooks.api.schemas.logs import ServerLogs
from renku_data_services.notebooks.config import GitProviderHelperProto, NotebooksConfig
from renku_data_services.notebooks.core_sessions import (
    patch_session,
    start_session,
    validate_session_post_request,
)
from renku_data_services.notebooks.errors.intermittent import AnonymousUserPatchError
from renku_data_services.notebooks.image_check import ImageCheckRepository
from renku_data_services.project.db import ProjectRepository, ProjectSessionSecretRepository
from renku_data_services.session.db import SessionRepository
from renku_data_services.storage.db import StorageRepository
from renku_data_services.users.db import UserRepo

logger = logging.getLogger(__name__)


@dataclass(kw_only=True)
class NotebooksBP(CustomBlueprint):
    """Handlers for manipulating notebooks."""

    authenticator: Authenticator
    nb_config: NotebooksConfig
    internal_gitlab_authenticator: base_models.Authenticator
    rp_repo: ResourcePoolRepository
    user_repo: UserRepo
    storage_repo: StorageRepository
    git_provider_helper: GitProviderHelperProto

    def version(self) -> BlueprintFactoryResponse:
        """Return notebook services version."""

        async def _version(_: Request) -> JSONResponse:
            return json(core.notebooks_info(self.nb_config))

        return "/notebooks/version", ["GET"], _version

    def user_servers(self) -> BlueprintFactoryResponse:
        """Return a JSON of running servers for the user."""

        @authenticate(self.authenticator)
        async def _user_servers(
            request: Request, user: AnonymousAPIUser | AuthenticatedAPIUser, **query_params: dict
        ) -> JSONResponse:
            filter_attrs = list(filter(lambda x: x[1] is not None, request.get_query_args()))
            filtered_servers = await core.user_servers(self.nb_config, user, filter_attrs)
            return core.serialize_v1_servers(filtered_servers, self.nb_config)

        return "/notebooks/servers", ["GET"], _user_servers

    def user_server(self) -> BlueprintFactoryResponse:
        """Returns a user server based on its ID."""

        @authenticate(self.authenticator)
        async def _user_server(
            request: Request, user: AnonymousAPIUser | AuthenticatedAPIUser, server_name: str
        ) -> JSONResponse:
            server = await core.user_server(self.nb_config, user, server_name)
            return core.serialize_v1_server(server, self.nb_config)

        return "/notebooks/servers/<server_name>", ["GET"], _user_server

    def launch_notebook(self) -> BlueprintFactoryResponse:
        """Start a renku session."""

        @authenticate_2(self.authenticator, self.internal_gitlab_authenticator)
        @validate(json=apispec.LaunchNotebookRequestOld)
        async def _launch_notebook(
            request: Request,
            user: AnonymousAPIUser | AuthenticatedAPIUser,
            internal_gitlab_user: APIUser,
            body: apispec.LaunchNotebookRequestOld,
        ) -> JSONResponse:
            server, status_code = await core.launch_notebook(
                self.nb_config,
                user,
                internal_gitlab_user,
                body,
                user_repo=self.user_repo,
                storage_repo=self.storage_repo,
                git_provider_helper=self.git_provider_helper,
            )
            return core.serialize_v1_server(server, self.nb_config, status_code)

        return "/notebooks/servers", ["POST"], _launch_notebook

    def patch_server(self) -> BlueprintFactoryResponse:
        """Patch a user server by name based on the query param."""

        @authenticate_2(self.authenticator, self.internal_gitlab_authenticator)
        @validate(json=apispec.PatchServerRequest)
        async def _patch_server(
            request: Request,
            user: AnonymousAPIUser | AuthenticatedAPIUser,
            internal_gitlab_user: APIUser,
            server_name: str,
            body: apispec.PatchServerRequest,
        ) -> JSONResponse:
            if isinstance(user, AnonymousAPIUser):
                raise AnonymousUserPatchError()

            manifest = await core.patch_server(self.nb_config, user, internal_gitlab_user, server_name, body)
            return core.serialize_v1_server(manifest, self.nb_config)

        return "/notebooks/servers/<server_name>", ["PATCH"], _patch_server

    def stop_server(self) -> BlueprintFactoryResponse:
        """Stop user server by name."""

        @authenticate(self.authenticator)
        async def _stop_server(
            _: Request, user: AnonymousAPIUser | AuthenticatedAPIUser, server_name: str
        ) -> HTTPResponse:
            try:
                await core.stop_server(self.nb_config, user, server_name)
            except errors.MissingResourceError as err:
                raise exceptions.NotFound(message=err.message) from err
            return HTTPResponse(status=204)

        return "/notebooks/servers/<server_name>", ["DELETE"], _stop_server

    def server_options(self) -> BlueprintFactoryResponse:
        """Return a set of configurable server options."""

        async def _server_options(request: Request) -> JSONResponse:
            return json(ServerOptionsEndpointResponse().dump(core.server_options(self.nb_config)))

        return "/notebooks/server_options", ["GET"], _server_options

    def server_logs(self) -> BlueprintFactoryResponse:
        """Return the logs of the running server."""

        @authenticate(self.authenticator)
        async def _server_logs(
            request: Request, user: AnonymousAPIUser | AuthenticatedAPIUser, server_name: str
        ) -> JSONResponse:
            args: dict[str, str | int] = request.get_args()
            max_lines = int(args.get("max_lines", 250))
            try:
                logs = await core.server_logs(self.nb_config, user, server_name, max_lines)
            except errors.MissingResourceError as err:
                raise exceptions.NotFound(message=err.message) from err
            return json(ServerLogs().dump(logs))

        return "/notebooks/logs/<server_name>", ["GET"], _server_logs

    def check_docker_image(self) -> BlueprintFactoryResponse:
        """Return the availability of the docker image."""

        @authenticate_2(self.authenticator, self.internal_gitlab_authenticator)
        @validate(query=apispec.NotebooksImagesGetParametersQuery)
        async def _check_docker_image(
            request: Request,
            user: AnonymousAPIUser | AuthenticatedAPIUser,
            internal_gitlab_user: APIUser,
            query: apispec.NotebooksImagesGetParametersQuery,
        ) -> HTTPResponse:
            image_url = request.get_args().get("image_url")
            if not isinstance(image_url, str):
                raise ValueError("required string of image url")

            status = 200 if await core.docker_image_exists(self.nb_config, image_url, internal_gitlab_user) else 404
            return HTTPResponse(status=status)

        return "/notebooks/images", ["GET"], _check_docker_image


@dataclass(kw_only=True)
class NotebooksNewBP(CustomBlueprint):
    """Handlers for manipulating notebooks for the new Amalthea operator."""

    authenticator: base_models.Authenticator
    internal_gitlab_authenticator: base_models.Authenticator
    nb_config: NotebooksConfig
    cluster_repo: ClusterRepository
    data_connector_repo: DataConnectorRepository
    data_connector_secret_repo: DataConnectorSecretRepository
    git_provider_helper: GitProviderHelperProto
    image_check_repo: ImageCheckRepository
    project_repo: ProjectRepository
    project_session_secret_repo: ProjectSessionSecretRepository
    rp_repo: ResourcePoolRepository
    session_repo: SessionRepository
    storage_repo: StorageRepository
    user_repo: UserRepo
    metrics: MetricsService

    def start(self) -> BlueprintFactoryResponse:
        """Start a session with the new operator."""

        @authenticate_2(self.authenticator, self.internal_gitlab_authenticator)
        @validate(json=apispec.SessionPostRequest)
        async def _handler(
            request: Request,
            user: AuthenticatedAPIUser | AnonymousAPIUser,
            internal_gitlab_user: APIUser,
            body: apispec.SessionPostRequest,
        ) -> JSONResponse:
            launch_request = validate_session_post_request(body=body)
            session, created = await start_session(
                request=request,
                launch_request=launch_request,
                user=user,
                internal_gitlab_user=internal_gitlab_user,
                nb_config=self.nb_config,
                git_provider_helper=self.git_provider_helper,
                cluster_repo=self.cluster_repo,
                data_connector_secret_repo=self.data_connector_secret_repo,
                project_repo=self.project_repo,
                project_session_secret_repo=self.project_session_secret_repo,
                rp_repo=self.rp_repo,
                session_repo=self.session_repo,
                user_repo=self.user_repo,
                metrics=self.metrics,
                image_check_repo=self.image_check_repo,
            )
            status = 201 if created else 200
            return json(session.as_apispec().model_dump(exclude_none=True, mode="json"), status)

        return "/sessions", ["POST"], _handler

    def get_all(self) -> BlueprintFactoryResponse:
        """Get all sessions for a user."""

        @authenticate(self.authenticator)
        async def _handler(_: Request, user: AuthenticatedAPIUser | AnonymousAPIUser) -> HTTPResponse:
            sessions = await self.nb_config.k8s_v2_client.list_sessions(user.id)
            output: list[dict] = []
            for session in sessions:
                output.append(session.as_apispec().model_dump(exclude_none=True, mode="json"))
            return json(output)

        return "/sessions", ["GET"], _handler

    def get_one(self) -> BlueprintFactoryResponse:
        """Get a specific session for a user."""

        @authenticate(self.authenticator)
        async def _handler(_: Request, user: AuthenticatedAPIUser | AnonymousAPIUser, session_id: str) -> HTTPResponse:
            session = await self.nb_config.k8s_v2_client.get_session(session_id, user.id)
            if session is None:
                raise errors.ValidationError(message=f"The session with ID {session_id} does not exist.", quiet=True)
            return json(session.as_apispec().model_dump(exclude_none=True, mode="json"))

        return "/sessions/<session_id>", ["GET"], _handler

    def delete(self) -> BlueprintFactoryResponse:
        """Fully delete a session with the new operator."""

        @authenticate(self.authenticator)
        async def _handler(_: Request, user: AuthenticatedAPIUser | AnonymousAPIUser, session_id: str) -> HTTPResponse:
            await self.nb_config.k8s_v2_client.delete_session(session_id, user.id)
            await self.metrics.session_stopped(user, metadata={"session_id": session_id})
            return empty()

        return "/sessions/<session_id>", ["DELETE"], _handler

    def patch(self) -> BlueprintFactoryResponse:
        """Patch a session."""

        @authenticate_2(self.authenticator, self.internal_gitlab_authenticator)
        @validate(json=apispec.SessionPatchRequest)
        async def _handler(
            _: Request,
            user: AuthenticatedAPIUser | AnonymousAPIUser,
            internal_gitlab_user: APIUser,
            session_id: str,
            body: apispec.SessionPatchRequest,
        ) -> HTTPResponse:
            new_session = await patch_session(
                body=body,
                session_id=session_id,
                user=user,
                internal_gitlab_user=internal_gitlab_user,
                nb_config=self.nb_config,
                git_provider_helper=self.git_provider_helper,
                project_repo=self.project_repo,
                project_session_secret_repo=self.project_session_secret_repo,
                rp_repo=self.rp_repo,
                session_repo=self.session_repo,
                metrics=self.metrics,
                image_check_repo=self.image_check_repo,
            )
            return json(new_session.as_apispec().model_dump(exclude_none=True, mode="json"))

        return "/sessions/<session_id>", ["PATCH"], _handler

    def logs(self) -> BlueprintFactoryResponse:
        """Get logs from the session."""

        @authenticate(self.authenticator)
        @validate(query=apispec.SessionsSessionIdLogsGetParametersQuery)
        async def _handler(
            _: Request,
            user: AuthenticatedAPIUser | AnonymousAPIUser,
            session_id: str,
            query: apispec.SessionsSessionIdLogsGetParametersQuery,
        ) -> HTTPResponse:
            logs = await self.nb_config.k8s_v2_client.get_session_logs(session_id, user.id, query.max_lines)
            return json(apispec.SessionLogsResponse.model_validate(logs).model_dump(exclude_none=True))

        return "/sessions/<session_id>/logs", ["GET"], _handler

    def check_docker_image(self) -> BlueprintFactoryResponse:
        """Return the availability of the docker image."""

        @authenticate_2(self.authenticator, self.internal_gitlab_authenticator)
        @validate(query=apispec.SessionsImagesGetParametersQuery)
        async def _check_docker_image(
            request: Request,
            user: AnonymousAPIUser | AuthenticatedAPIUser,
            internal_gitlab_user: APIUser,
            query: apispec.SessionsImagesGetParametersQuery,
        ) -> JSONResponse:
            image = Image.from_path(query.image_url)
            result = await self.image_check_repo.check_image(
                user=user,
                gitlab_user=internal_gitlab_user,
                image=image,
            )
            logger.info(f"Checked image {query.image_url}: {result}")
            conn = None
            if result.connection:
                match result.connection.status:
                    case ConnectionStatus.connected:
                        if result.error is not None:
                            status = apispec.ImageConnectionStatus.invalid_credentials
                        else:
                            status = apispec.ImageConnectionStatus.connected

                    case ConnectionStatus.pending:
                        status = apispec.ImageConnectionStatus.pending

                conn = apispec.ImageConnection(
                    id=str(result.connection.id), provider_id=result.connection.provider_id, status=status
                )

            provider: apispec.ImageProvider | None = None
            if result.client:
                provider = apispec.ImageProvider(
                    id=result.client.id, name=result.client.display_name, url=result.client.url
                )

            platforms = None
            if result.platforms:
                platforms = [apispec.ImagePlatform.model_validate(p) for p in result.platforms]

            resp = apispec.ImageCheckResponse(
                accessible=result.accessible, platforms=platforms, connection=conn, provider=provider
            )

            return json(resp.model_dump(exclude_none=True, mode="json"))

        return "/sessions/images", ["GET"], _check_docker_image

1	"""Notebooks service API."""
2
3	from dataclasses import dataclass	2✔
4
5	from sanic import Request, empty, exceptions, json	2✔
6	from sanic.response import HTTPResponse, JSONResponse	2✔
7	from sanic_ext import validate	2✔
8
9	from renku_data_services import base_models	2✔
10	from renku_data_services.app_config import logging	2✔
11	from renku_data_services.base_api.auth import authenticate, authenticate_2	2✔
12	from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint	2✔
13	from renku_data_services.base_models import AnonymousAPIUser, APIUser, AuthenticatedAPIUser, Authenticator	2✔
14	from renku_data_services.base_models.metrics import MetricsService	2✔
15	from renku_data_services.connected_services.models import ConnectionStatus	2✔
16	from renku_data_services.crc.db import ClusterRepository, ResourcePoolRepository	2✔
17	from renku_data_services.data_connectors.db import (	2✔
18	DataConnectorRepository,
19	DataConnectorSecretRepository,
20	)
21	from renku_data_services.errors import errors	2✔
22	from renku_data_services.notebooks import apispec, core	2✔
23	from renku_data_services.notebooks.api.classes.image import Image	2✔
24	from renku_data_services.notebooks.api.schemas.config_server_options import ServerOptionsEndpointResponse	2✔
25	from renku_data_services.notebooks.api.schemas.logs import ServerLogs	2✔
26	from renku_data_services.notebooks.config import GitProviderHelperProto, NotebooksConfig	2✔
27	from renku_data_services.notebooks.core_sessions import (	2✔
28	patch_session,
29	start_session,
30	validate_session_post_request,
31	)
32	from renku_data_services.notebooks.errors.intermittent import AnonymousUserPatchError	2✔
33	from renku_data_services.notebooks.image_check import ImageCheckRepository	2✔
34	from renku_data_services.project.db import ProjectRepository, ProjectSessionSecretRepository	2✔
35	from renku_data_services.session.db import SessionRepository	2✔
36	from renku_data_services.storage.db import StorageRepository	2✔
37	from renku_data_services.users.db import UserRepo	2✔
38
39	logger = logging.getLogger(__name__)	2✔
40
41
42	@dataclass(kw_only=True)	2✔
43	class NotebooksBP(CustomBlueprint):	2✔
44	"""Handlers for manipulating notebooks."""
45
46	authenticator: Authenticator	2✔
47	nb_config: NotebooksConfig	2✔
48	internal_gitlab_authenticator: base_models.Authenticator	2✔
49	rp_repo: ResourcePoolRepository	2✔
50	user_repo: UserRepo	2✔
51	storage_repo: StorageRepository	2✔
52	git_provider_helper: GitProviderHelperProto	2✔
53
54	def version(self) -> BlueprintFactoryResponse:	2✔
55	"""Return notebook services version."""
56
57	async def _version(_: Request) -> JSONResponse:	2✔
58	return json(core.notebooks_info(self.nb_config))	1✔
59
60	return "/notebooks/version", ["GET"], _version	2✔
61
62	def user_servers(self) -> BlueprintFactoryResponse:	2✔
63	"""Return a JSON of running servers for the user."""
64
65	@authenticate(self.authenticator)	2✔
66	async def _user_servers(	2✔
67	request: Request, user: AnonymousAPIUser \| AuthenticatedAPIUser, **query_params: dict
68	) -> JSONResponse:
69	filter_attrs = list(filter(lambda x: x[1] is not None, request.get_query_args()))	1✔
70	filtered_servers = await core.user_servers(self.nb_config, user, filter_attrs)	1✔
71	return core.serialize_v1_servers(filtered_servers, self.nb_config)	1✔
72
73	return "/notebooks/servers", ["GET"], _user_servers	2✔
74
75	def user_server(self) -> BlueprintFactoryResponse:	2✔
76	"""Returns a user server based on its ID."""
77
78	@authenticate(self.authenticator)	2✔
79	async def _user_server(	2✔
80	request: Request, user: AnonymousAPIUser \| AuthenticatedAPIUser, server_name: str
81	) -> JSONResponse:
82	server = await core.user_server(self.nb_config, user, server_name)	×
83	return core.serialize_v1_server(server, self.nb_config)	×
84
85	return "/notebooks/servers/<server_name>", ["GET"], _user_server	2✔
86
87	def launch_notebook(self) -> BlueprintFactoryResponse:	2✔
88	"""Start a renku session."""
89
90	@authenticate_2(self.authenticator, self.internal_gitlab_authenticator)	2✔
91	@validate(json=apispec.LaunchNotebookRequestOld)	2✔
92	async def _launch_notebook(	2✔
93	request: Request,
94	user: AnonymousAPIUser \| AuthenticatedAPIUser,
95	internal_gitlab_user: APIUser,
96	body: apispec.LaunchNotebookRequestOld,
97	) -> JSONResponse:
98	server, status_code = await core.launch_notebook(	1✔
99	self.nb_config,
100	user,
101	internal_gitlab_user,
102	body,
103	user_repo=self.user_repo,
104	storage_repo=self.storage_repo,
105	git_provider_helper=self.git_provider_helper,
106	)
107	return core.serialize_v1_server(server, self.nb_config, status_code)	1✔
108
109	return "/notebooks/servers", ["POST"], _launch_notebook	2✔
110
111	def patch_server(self) -> BlueprintFactoryResponse:	2✔
112	"""Patch a user server by name based on the query param."""
113
114	@authenticate_2(self.authenticator, self.internal_gitlab_authenticator)	2✔
115	@validate(json=apispec.PatchServerRequest)	2✔
116	async def _patch_server(	2✔
117	request: Request,
118	user: AnonymousAPIUser \| AuthenticatedAPIUser,
119	internal_gitlab_user: APIUser,
120	server_name: str,
121	body: apispec.PatchServerRequest,
122	) -> JSONResponse:
123	if isinstance(user, AnonymousAPIUser):	1✔
124	raise AnonymousUserPatchError()	×
125
126	manifest = await core.patch_server(self.nb_config, user, internal_gitlab_user, server_name, body)	1✔
127	return core.serialize_v1_server(manifest, self.nb_config)	1✔
128
129	return "/notebooks/servers/<server_name>", ["PATCH"], _patch_server	2✔
130
131	def stop_server(self) -> BlueprintFactoryResponse:	2✔
132	"""Stop user server by name."""
133
134	@authenticate(self.authenticator)	2✔
135	async def _stop_server(	2✔
136	_: Request, user: AnonymousAPIUser \| AuthenticatedAPIUser, server_name: str
137	) -> HTTPResponse:
138	try:	1✔
139	await core.stop_server(self.nb_config, user, server_name)	1✔
140	except errors.MissingResourceError as err:	×
141	raise exceptions.NotFound(message=err.message) from err	×
142	return HTTPResponse(status=204)	1✔
143
144	return "/notebooks/servers/<server_name>", ["DELETE"], _stop_server	2✔
145
146	def server_options(self) -> BlueprintFactoryResponse:	2✔
147	"""Return a set of configurable server options."""
148
149	async def _server_options(request: Request) -> JSONResponse:	2✔
150	return json(ServerOptionsEndpointResponse().dump(core.server_options(self.nb_config)))	1✔
151
152	return "/notebooks/server_options", ["GET"], _server_options	2✔
153
154	def server_logs(self) -> BlueprintFactoryResponse:	2✔
155	"""Return the logs of the running server."""
156
157	@authenticate(self.authenticator)	2✔
158	async def _server_logs(	2✔
159	request: Request, user: AnonymousAPIUser \| AuthenticatedAPIUser, server_name: str
160	) -> JSONResponse:
161	args: dict[str, str \| int] = request.get_args()	1✔
162	max_lines = int(args.get("max_lines", 250))	1✔
163	try:	1✔
164	logs = await core.server_logs(self.nb_config, user, server_name, max_lines)	1✔
165	except errors.MissingResourceError as err:	1✔
166	raise exceptions.NotFound(message=err.message) from err	1✔
167	return json(ServerLogs().dump(logs))	1✔
168
169	return "/notebooks/logs/<server_name>", ["GET"], _server_logs	2✔
170
171	def check_docker_image(self) -> BlueprintFactoryResponse:	2✔
172	"""Return the availability of the docker image."""
173
174	@authenticate_2(self.authenticator, self.internal_gitlab_authenticator)	2✔
175	@validate(query=apispec.NotebooksImagesGetParametersQuery)	2✔
176	async def _check_docker_image(	2✔
177	request: Request,
178	user: AnonymousAPIUser \| AuthenticatedAPIUser,
179	internal_gitlab_user: APIUser,
180	query: apispec.NotebooksImagesGetParametersQuery,
181	) -> HTTPResponse:
182	image_url = request.get_args().get("image_url")	1✔
183	if not isinstance(image_url, str):	1✔
184	raise ValueError("required string of image url")	×
185
186	status = 200 if await core.docker_image_exists(self.nb_config, image_url, internal_gitlab_user) else 404	1✔
187	return HTTPResponse(status=status)	1✔
188
189	return "/notebooks/images", ["GET"], _check_docker_image	2✔
190
191
192	@dataclass(kw_only=True)	2✔
193	class NotebooksNewBP(CustomBlueprint):	2✔
194	"""Handlers for manipulating notebooks for the new Amalthea operator."""
195
196	authenticator: base_models.Authenticator	2✔
197	internal_gitlab_authenticator: base_models.Authenticator	2✔
198	nb_config: NotebooksConfig	2✔
199	cluster_repo: ClusterRepository	2✔
200	data_connector_repo: DataConnectorRepository	2✔
201	data_connector_secret_repo: DataConnectorSecretRepository	2✔
202	git_provider_helper: GitProviderHelperProto	2✔
203	image_check_repo: ImageCheckRepository	2✔
204	project_repo: ProjectRepository	2✔
205	project_session_secret_repo: ProjectSessionSecretRepository	2✔
206	rp_repo: ResourcePoolRepository	2✔
207	session_repo: SessionRepository	2✔
208	storage_repo: StorageRepository	2✔
209	user_repo: UserRepo	2✔
210	metrics: MetricsService	2✔
211
212	def start(self) -> BlueprintFactoryResponse:	2✔
213	"""Start a session with the new operator."""
214
215	@authenticate_2(self.authenticator, self.internal_gitlab_authenticator)	2✔
216	@validate(json=apispec.SessionPostRequest)	2✔
217	async def _handler(	2✔
218	request: Request,
219	user: AuthenticatedAPIUser \| AnonymousAPIUser,
220	internal_gitlab_user: APIUser,
221	body: apispec.SessionPostRequest,
222	) -> JSONResponse:
223	launch_request = validate_session_post_request(body=body)	×
224	session, created = await start_session(	×
225	request=request,
226	launch_request=launch_request,
227	user=user,
228	internal_gitlab_user=internal_gitlab_user,
229	nb_config=self.nb_config,
230	git_provider_helper=self.git_provider_helper,
231	cluster_repo=self.cluster_repo,
232	data_connector_secret_repo=self.data_connector_secret_repo,
233	project_repo=self.project_repo,
234	project_session_secret_repo=self.project_session_secret_repo,
235	rp_repo=self.rp_repo,
236	session_repo=self.session_repo,
237	user_repo=self.user_repo,
238	metrics=self.metrics,
239	image_check_repo=self.image_check_repo,
240	)
241	status = 201 if created else 200	×
242	return json(session.as_apispec().model_dump(exclude_none=True, mode="json"), status)	×
243
244	return "/sessions", ["POST"], _handler	2✔
245
246	def get_all(self) -> BlueprintFactoryResponse:	2✔
247	"""Get all sessions for a user."""
248
249	@authenticate(self.authenticator)	2✔
250	async def _handler(_: Request, user: AuthenticatedAPIUser \| AnonymousAPIUser) -> HTTPResponse:	2✔
251	sessions = await self.nb_config.k8s_v2_client.list_sessions(user.id)	×
252	output: list[dict] = []	×
253	for session in sessions:	×
254	output.append(session.as_apispec().model_dump(exclude_none=True, mode="json"))	×
255	return json(output)	×
256
257	return "/sessions", ["GET"], _handler	2✔
258
259	def get_one(self) -> BlueprintFactoryResponse:	2✔
260	"""Get a specific session for a user."""
261
262	@authenticate(self.authenticator)	2✔
263	async def _handler(_: Request, user: AuthenticatedAPIUser \| AnonymousAPIUser, session_id: str) -> HTTPResponse:	2✔
264	session = await self.nb_config.k8s_v2_client.get_session(session_id, user.id)	×
265	if session is None:	×
266	raise errors.ValidationError(message=f"The session with ID {session_id} does not exist.", quiet=True)	×
267	return json(session.as_apispec().model_dump(exclude_none=True, mode="json"))	×
268
269	return "/sessions/<session_id>", ["GET"], _handler	2✔
270
271	def delete(self) -> BlueprintFactoryResponse:	2✔
272	"""Fully delete a session with the new operator."""
273
274	@authenticate(self.authenticator)	2✔
275	async def _handler(_: Request, user: AuthenticatedAPIUser \| AnonymousAPIUser, session_id: str) -> HTTPResponse:	2✔
276	await self.nb_config.k8s_v2_client.delete_session(session_id, user.id)	×
277	await self.metrics.session_stopped(user, metadata={"session_id": session_id})	×
278	return empty()	×
279
280	return "/sessions/<session_id>", ["DELETE"], _handler	2✔
281
282	def patch(self) -> BlueprintFactoryResponse:	2✔
283	"""Patch a session."""
284
285	@authenticate_2(self.authenticator, self.internal_gitlab_authenticator)	2✔
286	@validate(json=apispec.SessionPatchRequest)	2✔
287	async def _handler(	2✔
288	_: Request,
289	user: AuthenticatedAPIUser \| AnonymousAPIUser,
290	internal_gitlab_user: APIUser,
291	session_id: str,
292	body: apispec.SessionPatchRequest,
293	) -> HTTPResponse:
294	new_session = await patch_session(	×
295	body=body,
296	session_id=session_id,
297	user=user,
298	internal_gitlab_user=internal_gitlab_user,
299	nb_config=self.nb_config,
300	git_provider_helper=self.git_provider_helper,
301	project_repo=self.project_repo,
302	project_session_secret_repo=self.project_session_secret_repo,
303	rp_repo=self.rp_repo,
304	session_repo=self.session_repo,
305	metrics=self.metrics,
306	image_check_repo=self.image_check_repo,
307	)
308	return json(new_session.as_apispec().model_dump(exclude_none=True, mode="json"))	×
309
310	return "/sessions/<session_id>", ["PATCH"], _handler	2✔
311
312	def logs(self) -> BlueprintFactoryResponse:	2✔
313	"""Get logs from the session."""
314
315	@authenticate(self.authenticator)	2✔
316	@validate(query=apispec.SessionsSessionIdLogsGetParametersQuery)	2✔
317	async def _handler(	2✔
318	_: Request,
319	user: AuthenticatedAPIUser \| AnonymousAPIUser,
320	session_id: str,
321	query: apispec.SessionsSessionIdLogsGetParametersQuery,
322	) -> HTTPResponse:
323	logs = await self.nb_config.k8s_v2_client.get_session_logs(session_id, user.id, query.max_lines)	×
324	return json(apispec.SessionLogsResponse.model_validate(logs).model_dump(exclude_none=True))	×
325
326	return "/sessions/<session_id>/logs", ["GET"], _handler	2✔
327
328	def check_docker_image(self) -> BlueprintFactoryResponse:	2✔
329	"""Return the availability of the docker image."""
330
331	@authenticate_2(self.authenticator, self.internal_gitlab_authenticator)	2✔
332	@validate(query=apispec.SessionsImagesGetParametersQuery)	2✔
333	async def _check_docker_image(	2✔
334	request: Request,
335	user: AnonymousAPIUser \| AuthenticatedAPIUser,
336	internal_gitlab_user: APIUser,
337	query: apispec.SessionsImagesGetParametersQuery,
338	) -> JSONResponse:
339	image = Image.from_path(query.image_url)	×
NEW 340	result = await self.image_check_repo.check_image(	×
341	user=user,
342	gitlab_user=internal_gitlab_user,
343	image=image,
344	)
345	logger.info(f"Checked image {query.image_url}: {result}")	×
346	conn = None	×
347	if result.connection:	×
348	match result.connection.status:	×
349	case ConnectionStatus.connected:	×
350	if result.error is not None:	×
351	status = apispec.ImageConnectionStatus.invalid_credentials	×
352	else:
353	status = apispec.ImageConnectionStatus.connected	×
354
355	case ConnectionStatus.pending:	×
356	status = apispec.ImageConnectionStatus.pending	×
357
358	conn = apispec.ImageConnection(	×
359	id=str(result.connection.id), provider_id=result.connection.provider_id, status=status
360	)
361
362	provider: apispec.ImageProvider \| None = None	×
363	if result.client:	×
364	provider = apispec.ImageProvider(	×
365	id=result.client.id, name=result.client.display_name, url=result.client.url
366	)
367
NEW 368	platforms = None	×
NEW 369	if result.platforms:	×
NEW 370	platforms = [apispec.ImagePlatform.model_validate(p) for p in result.platforms]	×
371
NEW 372	resp = apispec.ImageCheckResponse(	×
373	accessible=result.accessible, platforms=platforms, connection=conn, provider=provider
374	)
375
376	return json(resp.model_dump(exclude_none=True, mode="json"))	×
377
378	return "/sessions/images", ["GET"], _check_docker_image	2✔

SwissDataScienceCenter / renku-data-services / 19145569596

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous