12047437567

Committed 27 Nov 2024 09:31AM UTC coverage: 86.006% (+0.1%) from 85.882%

Build # 12047437567

Build Type

Pull #545

github

Committed by

web-flow

Commit Message

Merge c434d43ab into 7ae3af62e

Pull Request Pull Request #545: feat!: add support for session secrets

Run Details

216 of 222 new or added lines in 10 files covered. (97.3%)

5 existing lines in 3 files now uncovered.

14682 of 17071 relevant lines covered (86.01%)

1.52 hits per line

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

97.46

/components/renku_data_services/project/blueprints.py

"""Project blueprint."""

from dataclasses import dataclass
from typing import Any

from sanic import HTTPResponse, Request
from sanic.response import JSONResponse
from sanic_ext import validate
from ulid import ULID

import renku_data_services.base_models as base_models
from renku_data_services.authz.models import Member, Role, Visibility
from renku_data_services.base_api.auth import (
    authenticate,
    only_authenticated,
    validate_path_user_id,
)
from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint
from renku_data_services.base_api.etag import extract_if_none_match, if_match_required
from renku_data_services.base_api.misc import validate_body_root_model, validate_query
from renku_data_services.base_api.pagination import PaginationRequest, paginate
from renku_data_services.base_models.validation import validate_and_dump, validated_json
from renku_data_services.data_connectors.db import DataConnectorProjectLinkRepository
from renku_data_services.errors import errors
from renku_data_services.project import apispec
from renku_data_services.project import models as project_models
from renku_data_services.project.core import (
    copy_project,
    validate_project_patch,
    validate_session_secret_slot_patch,
    validate_unsaved_session_secret_slot,
)
from renku_data_services.project.db import ProjectMemberRepository, ProjectRepository, ProjectSessionSecretRepository
from renku_data_services.session.db import SessionRepository
from renku_data_services.users.db import UserRepo


@dataclass(kw_only=True)
class ProjectsBP(CustomBlueprint):
    """Handlers for manipulating projects."""

    project_repo: ProjectRepository
    project_member_repo: ProjectMemberRepository
    user_repo: UserRepo
    authenticator: base_models.Authenticator
    session_repo: SessionRepository
    data_connector_to_project_link_repo: DataConnectorProjectLinkRepository

    def get_all(self) -> BlueprintFactoryResponse:
        """List all projects."""

        @authenticate(self.authenticator)
        @validate_query(query=apispec.ProjectGetQuery)
        @paginate
        async def _get_all(
            _: Request, user: base_models.APIUser, pagination: PaginationRequest, query: apispec.ProjectGetQuery
        ) -> tuple[list[dict[str, Any]], int]:
            projects, total_num = await self.project_repo.get_projects(
                user=user, pagination=pagination, namespace=query.namespace, direct_member=query.direct_member
            )
            return [validate_and_dump(apispec.Project, self._dump_project(p)) for p in projects], total_num

        return "/projects", ["GET"], _get_all

    def post(self) -> BlueprintFactoryResponse:
        """Create a new project."""

        @authenticate(self.authenticator)
        @only_authenticated
        @validate(json=apispec.ProjectPost)
        async def _post(_: Request, user: base_models.APIUser, body: apispec.ProjectPost) -> JSONResponse:
            keywords = [kw.root for kw in body.keywords] if body.keywords is not None else []
            visibility = Visibility.PRIVATE if body.visibility is None else Visibility(body.visibility.value)
            project = project_models.UnsavedProject(
                name=body.name,
                namespace=body.namespace,
                slug=body.slug or base_models.Slug.from_name(body.name).value,
                description=body.description,
                repositories=body.repositories or [],
                created_by=user.id,  # type: ignore[arg-type]
                visibility=visibility,
                keywords=keywords,
                documentation=body.documentation,
            )
            result = await self.project_repo.insert_project(user, project)
            return validated_json(apispec.Project, self._dump_project(result), status=201)

        return "/projects", ["POST"], _post

    def copy(self) -> BlueprintFactoryResponse:
        """Create a new project by copying it from a template project."""

        @authenticate(self.authenticator)
        @only_authenticated
        @validate(json=apispec.ProjectPost)
        async def _copy(
            _: Request, user: base_models.APIUser, project_id: ULID, body: apispec.ProjectPost
        ) -> JSONResponse:
            project = await copy_project(
                project_id=project_id,
                user=user,
                name=body.name,
                namespace=body.namespace,
                slug=body.slug,
                description=body.description,
                repositories=body.repositories,
                visibility=Visibility(body.visibility.value) if body.visibility is not None else None,
                keywords=[kw.root for kw in body.keywords] if body.keywords is not None else [],
                project_repo=self.project_repo,
                session_repo=self.session_repo,
                data_connector_to_project_link_repo=self.data_connector_to_project_link_repo,
            )
            return validated_json(apispec.Project, self._dump_project(project), status=201)

        return "/projects/<project_id:ulid>/copies", ["POST"], _copy

    def get_all_copies(self) -> BlueprintFactoryResponse:
        """Get all copies of a specific project that the user has access to."""

        @authenticate(self.authenticator)
        @only_authenticated
        @validate(query=apispec.ProjectsProjectIdCopiesGetParametersQuery)
        async def _get_all_copies(
            _: Request,
            user: base_models.APIUser,
            project_id: ULID,
            query: apispec.ProjectsProjectIdCopiesGetParametersQuery,
        ) -> JSONResponse:
            projects = await self.project_repo.get_all_copied_projects(
                user=user, project_id=project_id, only_writable=query.writable
            )
            projects_dump = [self._dump_project(p) for p in projects]
            return validated_json(apispec.ProjectsList, projects_dump)

        return "/projects/<project_id:ulid>/copies", ["GET"], _get_all_copies

    def get_one(self) -> BlueprintFactoryResponse:
        """Get a specific project."""

        @authenticate(self.authenticator)
        @extract_if_none_match
        @validate_query(query=apispec.ProjectsProjectIdGetParametersQuery)
        async def _get_one(
            _: Request,
            user: base_models.APIUser,
            project_id: ULID,
            etag: str | None,
            query: apispec.ProjectsProjectIdGetParametersQuery,
        ) -> JSONResponse | HTTPResponse:
            with_documentation = query.with_documentation is True
            project = await self.project_repo.get_project(
                user=user, project_id=project_id, with_documentation=with_documentation
            )

            if project.etag is not None and project.etag == etag:
                return HTTPResponse(status=304)

            headers = {"ETag": project.etag} if project.etag is not None else None
            return validated_json(
                apispec.Project, self._dump_project(project, with_documentation=with_documentation), headers=headers
            )

        return "/projects/<project_id:ulid>", ["GET"], _get_one

    def get_one_by_namespace_slug(self) -> BlueprintFactoryResponse:
        """Get a specific project by namespace/slug."""

        @authenticate(self.authenticator)
        @extract_if_none_match
        @validate_query(query=apispec.NamespacesNamespaceProjectsSlugGetParametersQuery)
        async def _get_one_by_namespace_slug(
            _: Request,
            user: base_models.APIUser,
            namespace: str,
            slug: str,
            etag: str | None,
            query: apispec.NamespacesNamespaceProjectsSlugGetParametersQuery,
        ) -> JSONResponse | HTTPResponse:
            with_documentation = query.with_documentation is True
            project = await self.project_repo.get_project_by_namespace_slug(
                user=user, namespace=namespace, slug=slug, with_documentation=with_documentation
            )

            if project.etag is not None and project.etag == etag:
                return HTTPResponse(status=304)

            headers = {"ETag": project.etag} if project.etag is not None else None
            return validated_json(
                apispec.Project,
                self._dump_project(project, with_documentation=with_documentation),
                headers=headers,
            )

        return "/namespaces/<namespace>/projects/<slug:renku_slug>", ["GET"], _get_one_by_namespace_slug

    def delete(self) -> BlueprintFactoryResponse:
        """Delete a specific project."""

        @authenticate(self.authenticator)
        @only_authenticated
        async def _delete(_: Request, user: base_models.APIUser, project_id: ULID) -> HTTPResponse:
            await self.project_repo.delete_project(user=user, project_id=project_id)
            return HTTPResponse(status=204)

        return "/projects/<project_id:ulid>", ["DELETE"], _delete

    def patch(self) -> BlueprintFactoryResponse:
        """Partially update a specific project."""

        @authenticate(self.authenticator)
        @only_authenticated
        @if_match_required
        @validate(json=apispec.ProjectPatch)
        async def _patch(
            _: Request, user: base_models.APIUser, project_id: ULID, body: apispec.ProjectPatch, etag: str
        ) -> JSONResponse:
            project_patch = validate_project_patch(body)
            project_update = await self.project_repo.update_project(
                user=user, project_id=project_id, etag=etag, patch=project_patch
            )

            if not isinstance(project_update, project_models.ProjectUpdate):
                raise errors.ProgrammingError(
                    message="Expected the result of a project update to be ProjectUpdate but instead "
                    f"got {type(project_update)}"
                )

            updated_project = project_update.new
            return validated_json(apispec.Project, self._dump_project(updated_project))

        return "/projects/<project_id:ulid>", ["PATCH"], _patch

    def get_all_members(self) -> BlueprintFactoryResponse:
        """List all project members."""

        @authenticate(self.authenticator)
        async def _get_all_members(_: Request, user: base_models.APIUser, project_id: ULID) -> JSONResponse:
            members = await self.project_member_repo.get_members(user, project_id)

            users = []

            for member in members:
                user_id = member.user_id
                user_info = await self.user_repo.get_user(id=user_id)
                if not user_info:
                    raise errors.MissingResourceError(message=f"The user with ID {user_id} cannot be found.")
                namespace_info = user_info.namespace

                user_with_id = apispec.ProjectMemberResponse(
                    id=user_id,
                    namespace=namespace_info.slug,
                    first_name=user_info.first_name,
                    last_name=user_info.last_name,
                    role=apispec.Role(member.role.value),
                ).model_dump(exclude_none=True, mode="json")
                users.append(user_with_id)

            return validated_json(apispec.ProjectMemberListResponse, users)

        return "/projects/<project_id:ulid>/members", ["GET"], _get_all_members

    def update_members(self) -> BlueprintFactoryResponse:
        """Update or add project members."""

        @authenticate(self.authenticator)
        @validate_body_root_model(json=apispec.ProjectMemberListPatchRequest)
        async def _update_members(
            _: Request, user: base_models.APIUser, project_id: ULID, body: apispec.ProjectMemberListPatchRequest
        ) -> HTTPResponse:
            members = [Member(Role(i.role.value), i.id, project_id) for i in body.root]
            await self.project_member_repo.update_members(user, project_id, members)
            return HTTPResponse(status=200)

        return "/projects/<project_id:ulid>/members", ["PATCH"], _update_members

    def delete_member(self) -> BlueprintFactoryResponse:
        """Delete a specific project."""

        @authenticate(self.authenticator)
        @validate_path_user_id
        async def _delete_member(
            _: Request, user: base_models.APIUser, project_id: ULID, member_id: str
        ) -> HTTPResponse:
            await self.project_member_repo.delete_members(user, project_id, [member_id])
            return HTTPResponse(status=204)

        return "/projects/<project_id:ulid>/members/<member_id>", ["DELETE"], _delete_member

    def get_permissions(self) -> BlueprintFactoryResponse:
        """Get the permissions of the current user on the project."""

        @authenticate(self.authenticator)
        async def _get_permissions(_: Request, user: base_models.APIUser, project_id: ULID) -> JSONResponse:
            permissions = await self.project_repo.get_project_permissions(user=user, project_id=project_id)
            return validated_json(apispec.ProjectPermissions, permissions)

        return "/projects/<project_id:ulid>/permissions", ["GET"], _get_permissions

    @staticmethod
    def _dump_project(project: project_models.Project, with_documentation: bool = False) -> dict[str, Any]:
        """Dumps a project for API responses."""
        result = dict(
            id=project.id,
            name=project.name,
            namespace=project.namespace.slug,
            slug=project.slug,
            creation_date=project.creation_date.isoformat(),
            created_by=project.created_by,
            updated_at=project.updated_at.isoformat() if project.updated_at else None,
            repositories=project.repositories,
            visibility=project.visibility.value,
            description=project.description,
            etag=project.etag,
            keywords=project.keywords or [],
            template_id=project.template_id,
            is_template=project.is_template,
        )
        if with_documentation:
            result = dict(result, documentation=project.documentation)
        return result


@dataclass(kw_only=True)
class ProjectSessionSecretBP(CustomBlueprint):
    """Handlers for manipulating session secrets in a project."""

    session_secret_repo: ProjectSessionSecretRepository
    authenticator: base_models.Authenticator

    def get_session_secret_slots(self) -> BlueprintFactoryResponse:
        """Get the session secret slots of a project."""

        @authenticate(self.authenticator)
        async def _get_session_secret_slots(_: Request, user: base_models.APIUser, project_id: ULID) -> JSONResponse:
            secret_slots = await self.session_secret_repo.get_all_session_secret_slots_from_project(
                user=user, project_id=project_id
            )
            return validated_json(apispec.SessionSecretSlotList, secret_slots)

        return "/projects/<project_id:ulid>/session_secret_slots", ["GET"], _get_session_secret_slots

    def post_session_secret_slot(self) -> BlueprintFactoryResponse:
        """Create a new session secret slot on a project."""

        @authenticate(self.authenticator)
        @only_authenticated
        @validate(json=apispec.SessionSecretSlotPost)
        async def _post_session_secret_slot(
            _: Request, user: base_models.APIUser, body: apispec.SessionSecretSlotPost
        ) -> JSONResponse:
            unsaved_secret_slot = validate_unsaved_session_secret_slot(body)
            secret_slot = await self.session_secret_repo.insert_session_secret_slot(
                user=user, secret_slot=unsaved_secret_slot
            )
            return validated_json(apispec.SessionSecretSlot, secret_slot, status=201)

        return "/session_secret_slots", ["POST"], _post_session_secret_slot

    def get_session_secret_slot(self) -> BlueprintFactoryResponse:
        """Get the details of a session secret slot."""

        @authenticate(self.authenticator)
        @extract_if_none_match
        async def _get_session_secret_slot(
            _: Request, user: base_models.APIUser, slot_id: ULID, etag: str | None
        ) -> HTTPResponse:
            secret_slot = await self.session_secret_repo.get_session_secret_slot(user=user, slot_id=slot_id)

            if secret_slot.etag == etag:
                return HTTPResponse(status=304)

            return validated_json(apispec.SessionSecretSlot, secret_slot)

        return "/session_secret_slots/<slot_id:ulid>", ["GET"], _get_session_secret_slot

    def patch_session_secret_slot(self) -> BlueprintFactoryResponse:
        """Update specific fields of an existing session secret slot."""

        @authenticate(self.authenticator)
        @only_authenticated
        @if_match_required
        @validate(json=apispec.SessionSecretSlotPatch)
        async def _patch_session_secret_slot(
            _: Request,
            user: base_models.APIUser,
            slot_id: ULID,
            body: apispec.SessionSecretSlotPatch,
            etag: str,
        ) -> JSONResponse:
            secret_slot_patch = validate_session_secret_slot_patch(body)
            secret_slot = await self.session_secret_repo.update_session_secret_slot(
                user=user, slot_id=slot_id, patch=secret_slot_patch, etag=etag
            )
            return validated_json(apispec.SessionSecretSlot, secret_slot)

        return "/session_secret_slots/<slot_id:ulid>", ["PATCH"], _patch_session_secret_slot

    def delete_session_secret_slot(self) -> BlueprintFactoryResponse:
        """Remove a session secret slot."""

        @authenticate(self.authenticator)
        @only_authenticated
        async def _delete_session_secret_slot(_: Request, user: base_models.APIUser, slot_id: ULID) -> HTTPResponse:
            await self.session_secret_repo.delete_session_secret_slot(user=user, slot_id=slot_id)
            return HTTPResponse(status=204)

        return "/session_secret_slots/<slot_id:ulid>", ["DELETE"], _delete_session_secret_slot

1	"""Project blueprint."""
2
3	from dataclasses import dataclass	2✔
4	from typing import Any	2✔
5
6	from sanic import HTTPResponse, Request	2✔
7	from sanic.response import JSONResponse	2✔
8	from sanic_ext import validate	2✔
9	from ulid import ULID	2✔
10
11	import renku_data_services.base_models as base_models	2✔
12	from renku_data_services.authz.models import Member, Role, Visibility	2✔
13	from renku_data_services.base_api.auth import (	2✔
14	authenticate,
15	only_authenticated,
16	validate_path_user_id,
17	)
18	from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint	2✔
19	from renku_data_services.base_api.etag import extract_if_none_match, if_match_required	2✔
20	from renku_data_services.base_api.misc import validate_body_root_model, validate_query	2✔
21	from renku_data_services.base_api.pagination import PaginationRequest, paginate	2✔
22	from renku_data_services.base_models.validation import validate_and_dump, validated_json	2✔
23	from renku_data_services.data_connectors.db import DataConnectorProjectLinkRepository	2✔
24	from renku_data_services.errors import errors	2✔
25	from renku_data_services.project import apispec	2✔
26	from renku_data_services.project import models as project_models	2✔
27	from renku_data_services.project.core import (	2✔
28	copy_project,
29	validate_project_patch,
30	validate_session_secret_slot_patch,
31	validate_unsaved_session_secret_slot,
32	)
33	from renku_data_services.project.db import ProjectMemberRepository, ProjectRepository, ProjectSessionSecretRepository	2✔
34	from renku_data_services.session.db import SessionRepository	2✔
35	from renku_data_services.users.db import UserRepo	2✔
36
37
38	@dataclass(kw_only=True)	2✔
39	class ProjectsBP(CustomBlueprint):	2✔
40	"""Handlers for manipulating projects."""
41
42	project_repo: ProjectRepository	2✔
43	project_member_repo: ProjectMemberRepository	2✔
44	user_repo: UserRepo	2✔
45	authenticator: base_models.Authenticator	2✔
46	session_repo: SessionRepository	2✔
47	data_connector_to_project_link_repo: DataConnectorProjectLinkRepository	2✔
48
49	def get_all(self) -> BlueprintFactoryResponse:	2✔
50	"""List all projects."""
51
52	@authenticate(self.authenticator)	2✔
53	@validate_query(query=apispec.ProjectGetQuery)	2✔
54	@paginate	2✔
55	async def _get_all(	2✔
56	_: Request, user: base_models.APIUser, pagination: PaginationRequest, query: apispec.ProjectGetQuery
57	) -> tuple[list[dict[str, Any]], int]:
58	projects, total_num = await self.project_repo.get_projects(	2✔
59	user=user, pagination=pagination, namespace=query.namespace, direct_member=query.direct_member
60	)
61	return [validate_and_dump(apispec.Project, self._dump_project(p)) for p in projects], total_num	2✔
62
63	return "/projects", ["GET"], _get_all	2✔
64
65	def post(self) -> BlueprintFactoryResponse:	2✔
66	"""Create a new project."""
67
68	@authenticate(self.authenticator)	2✔
69	@only_authenticated	2✔
70	@validate(json=apispec.ProjectPost)	2✔
71	async def _post(_: Request, user: base_models.APIUser, body: apispec.ProjectPost) -> JSONResponse:	2✔
72	keywords = [kw.root for kw in body.keywords] if body.keywords is not None else []	2✔
73	visibility = Visibility.PRIVATE if body.visibility is None else Visibility(body.visibility.value)	2✔
74	project = project_models.UnsavedProject(	2✔
75	name=body.name,
76	namespace=body.namespace,
77	slug=body.slug or base_models.Slug.from_name(body.name).value,
78	description=body.description,
79	repositories=body.repositories or [],
80	created_by=user.id, # type: ignore[arg-type]
81	visibility=visibility,
82	keywords=keywords,
83	documentation=body.documentation,
84	)
85	result = await self.project_repo.insert_project(user, project)	2✔
86	return validated_json(apispec.Project, self._dump_project(result), status=201)	1✔
87
88	return "/projects", ["POST"], _post	2✔
89
90	def copy(self) -> BlueprintFactoryResponse:	2✔
91	"""Create a new project by copying it from a template project."""
92
93	@authenticate(self.authenticator)	2✔
94	@only_authenticated	2✔
95	@validate(json=apispec.ProjectPost)	2✔
96	async def _copy(	2✔
97	_: Request, user: base_models.APIUser, project_id: ULID, body: apispec.ProjectPost
98	) -> JSONResponse:
99	project = await copy_project(	2✔
100	project_id=project_id,
101	user=user,
102	name=body.name,
103	namespace=body.namespace,
104	slug=body.slug,
105	description=body.description,
106	repositories=body.repositories,
107	visibility=Visibility(body.visibility.value) if body.visibility is not None else None,
108	keywords=[kw.root for kw in body.keywords] if body.keywords is not None else [],
109	project_repo=self.project_repo,
110	session_repo=self.session_repo,
111	data_connector_to_project_link_repo=self.data_connector_to_project_link_repo,
112	)
113	return validated_json(apispec.Project, self._dump_project(project), status=201)	1✔
114
115	return "/projects/<project_id:ulid>/copies", ["POST"], _copy	2✔
116
117	def get_all_copies(self) -> BlueprintFactoryResponse:	2✔
118	"""Get all copies of a specific project that the user has access to."""
119
120	@authenticate(self.authenticator)	2✔
121	@only_authenticated	2✔
122	@validate(query=apispec.ProjectsProjectIdCopiesGetParametersQuery)	2✔
123	async def _get_all_copies(	2✔
124	_: Request,
125	user: base_models.APIUser,
126	project_id: ULID,
127	query: apispec.ProjectsProjectIdCopiesGetParametersQuery,
128	) -> JSONResponse:
129	projects = await self.project_repo.get_all_copied_projects(	2✔
130	user=user, project_id=project_id, only_writable=query.writable
131	)
132	projects_dump = [self._dump_project(p) for p in projects]	1✔
133	return validated_json(apispec.ProjectsList, projects_dump)	1✔
134
135	return "/projects/<project_id:ulid>/copies", ["GET"], _get_all_copies	2✔
136
137	def get_one(self) -> BlueprintFactoryResponse:	2✔
138	"""Get a specific project."""
139
140	@authenticate(self.authenticator)	2✔
141	@extract_if_none_match	2✔
142	@validate_query(query=apispec.ProjectsProjectIdGetParametersQuery)	2✔
143	async def _get_one(	2✔
144	_: Request,
145	user: base_models.APIUser,
146	project_id: ULID,
147	etag: str \| None,
148	query: apispec.ProjectsProjectIdGetParametersQuery,
149	) -> JSONResponse \| HTTPResponse:
150	with_documentation = query.with_documentation is True	2✔
151	project = await self.project_repo.get_project(	2✔
152	user=user, project_id=project_id, with_documentation=with_documentation
153	)
154
155	if project.etag is not None and project.etag == etag:	1✔
156	return HTTPResponse(status=304)	×
157
158	headers = {"ETag": project.etag} if project.etag is not None else None	1✔
159	return validated_json(	1✔
160	apispec.Project, self._dump_project(project, with_documentation=with_documentation), headers=headers
161	)
162
163	return "/projects/<project_id:ulid>", ["GET"], _get_one	2✔
164
165	def get_one_by_namespace_slug(self) -> BlueprintFactoryResponse:	2✔
166	"""Get a specific project by namespace/slug."""
167
168	@authenticate(self.authenticator)	2✔
169	@extract_if_none_match	2✔
170	@validate_query(query=apispec.NamespacesNamespaceProjectsSlugGetParametersQuery)	2✔
171	async def _get_one_by_namespace_slug(	2✔
172	_: Request,
173	user: base_models.APIUser,
174	namespace: str,
175	slug: str,
176	etag: str \| None,
177	query: apispec.NamespacesNamespaceProjectsSlugGetParametersQuery,
178	) -> JSONResponse \| HTTPResponse:
179	with_documentation = query.with_documentation is True	2✔
180	project = await self.project_repo.get_project_by_namespace_slug(	2✔
181	user=user, namespace=namespace, slug=slug, with_documentation=with_documentation
182	)
183
184	if project.etag is not None and project.etag == etag:	1✔
185	return HTTPResponse(status=304)	×
186
187	headers = {"ETag": project.etag} if project.etag is not None else None	1✔
188	return validated_json(	1✔
189	apispec.Project,
190	self._dump_project(project, with_documentation=with_documentation),
191	headers=headers,
192	)
193
194	return "/namespaces/<namespace>/projects/<slug:renku_slug>", ["GET"], _get_one_by_namespace_slug	2✔
195
196	def delete(self) -> BlueprintFactoryResponse:	2✔
197	"""Delete a specific project."""
198
199	@authenticate(self.authenticator)	2✔
200	@only_authenticated	2✔
201	async def _delete(_: Request, user: base_models.APIUser, project_id: ULID) -> HTTPResponse:	2✔
202	await self.project_repo.delete_project(user=user, project_id=project_id)	2✔
203	return HTTPResponse(status=204)	1✔
204
205	return "/projects/<project_id:ulid>", ["DELETE"], _delete	2✔
206
207	def patch(self) -> BlueprintFactoryResponse:	2✔
208	"""Partially update a specific project."""
209
210	@authenticate(self.authenticator)	2✔
211	@only_authenticated	2✔
212	@if_match_required	2✔
213	@validate(json=apispec.ProjectPatch)	2✔
214	async def _patch(	2✔
215	_: Request, user: base_models.APIUser, project_id: ULID, body: apispec.ProjectPatch, etag: str
216	) -> JSONResponse:
217	project_patch = validate_project_patch(body)	2✔
218	project_update = await self.project_repo.update_project(	2✔
219	user=user, project_id=project_id, etag=etag, patch=project_patch
220	)
221
222	if not isinstance(project_update, project_models.ProjectUpdate):	1✔
223	raise errors.ProgrammingError(	×
224	message="Expected the result of a project update to be ProjectUpdate but instead "
225	f"got {type(project_update)}"
226	)
227
228	updated_project = project_update.new	1✔
229	return validated_json(apispec.Project, self._dump_project(updated_project))	1✔
230
231	return "/projects/<project_id:ulid>", ["PATCH"], _patch	2✔
232
233	def get_all_members(self) -> BlueprintFactoryResponse:	2✔
234	"""List all project members."""
235
236	@authenticate(self.authenticator)	2✔
237	async def _get_all_members(_: Request, user: base_models.APIUser, project_id: ULID) -> JSONResponse:	2✔
238	members = await self.project_member_repo.get_members(user, project_id)	2✔
239
240	users = []	1✔
241
242	for member in members:	1✔
243	user_id = member.user_id	1✔
244	user_info = await self.user_repo.get_user(id=user_id)	1✔
245	if not user_info:	1✔
246	raise errors.MissingResourceError(message=f"The user with ID {user_id} cannot be found.")	×
247	namespace_info = user_info.namespace	1✔
248
249	user_with_id = apispec.ProjectMemberResponse(	1✔
250	id=user_id,
251	namespace=namespace_info.slug,
252	first_name=user_info.first_name,
253	last_name=user_info.last_name,
254	role=apispec.Role(member.role.value),
255	).model_dump(exclude_none=True, mode="json")
256	users.append(user_with_id)	1✔
257
258	return validated_json(apispec.ProjectMemberListResponse, users)	1✔
259
260	return "/projects/<project_id:ulid>/members", ["GET"], _get_all_members	2✔
261
262	def update_members(self) -> BlueprintFactoryResponse:	2✔
263	"""Update or add project members."""
264
265	@authenticate(self.authenticator)	2✔
266	@validate_body_root_model(json=apispec.ProjectMemberListPatchRequest)	2✔
267	async def _update_members(	2✔
268	_: Request, user: base_models.APIUser, project_id: ULID, body: apispec.ProjectMemberListPatchRequest
269	) -> HTTPResponse:
270	members = [Member(Role(i.role.value), i.id, project_id) for i in body.root]	2✔
271	await self.project_member_repo.update_members(user, project_id, members)	2✔
272	return HTTPResponse(status=200)	1✔
273
274	return "/projects/<project_id:ulid>/members", ["PATCH"], _update_members	2✔
275
276	def delete_member(self) -> BlueprintFactoryResponse:	2✔
277	"""Delete a specific project."""
278
279	@authenticate(self.authenticator)	2✔
280	@validate_path_user_id	2✔
281	async def _delete_member(	2✔
282	_: Request, user: base_models.APIUser, project_id: ULID, member_id: str
283	) -> HTTPResponse:
284	await self.project_member_repo.delete_members(user, project_id, [member_id])	2✔
285	return HTTPResponse(status=204)	1✔
286
287	return "/projects/<project_id:ulid>/members/<member_id>", ["DELETE"], _delete_member	2✔
288
289	def get_permissions(self) -> BlueprintFactoryResponse:	2✔
290	"""Get the permissions of the current user on the project."""
291
292	@authenticate(self.authenticator)	2✔
293	async def _get_permissions(_: Request, user: base_models.APIUser, project_id: ULID) -> JSONResponse:	2✔
294	permissions = await self.project_repo.get_project_permissions(user=user, project_id=project_id)	2✔
295	return validated_json(apispec.ProjectPermissions, permissions)	1✔
296
297	return "/projects/<project_id:ulid>/permissions", ["GET"], _get_permissions	2✔
298
299	@staticmethod	2✔
300	def _dump_project(project: project_models.Project, with_documentation: bool = False) -> dict[str, Any]:	2✔
301	"""Dumps a project for API responses."""
302	result = dict(	1✔
303	id=project.id,
304	name=project.name,
305	namespace=project.namespace.slug,
306	slug=project.slug,
307	creation_date=project.creation_date.isoformat(),
308	created_by=project.created_by,
309	updated_at=project.updated_at.isoformat() if project.updated_at else None,
310	repositories=project.repositories,
311	visibility=project.visibility.value,
312	description=project.description,
313	etag=project.etag,
314	keywords=project.keywords or [],
315	template_id=project.template_id,
316	is_template=project.is_template,
317	)
318	if with_documentation:	1✔
319	result = dict(result, documentation=project.documentation)	1✔
320	return result	1✔
321
322
323	@dataclass(kw_only=True)	2✔
324	class ProjectSessionSecretBP(CustomBlueprint):	2✔
325	"""Handlers for manipulating session secrets in a project."""
326
327	session_secret_repo: ProjectSessionSecretRepository	2✔
328	authenticator: base_models.Authenticator	2✔
329
330	def get_session_secret_slots(self) -> BlueprintFactoryResponse:	2✔
331	"""Get the session secret slots of a project."""
332
333	@authenticate(self.authenticator)	2✔
334	async def _get_session_secret_slots(_: Request, user: base_models.APIUser, project_id: ULID) -> JSONResponse:	2✔
335	secret_slots = await self.session_secret_repo.get_all_session_secret_slots_from_project(	2✔
336	user=user, project_id=project_id
337	)
338	return validated_json(apispec.SessionSecretSlotList, secret_slots)	1✔
339
340	return "/projects/<project_id:ulid>/session_secret_slots", ["GET"], _get_session_secret_slots	2✔
341
342	def post_session_secret_slot(self) -> BlueprintFactoryResponse:	2✔
343	"""Create a new session secret slot on a project."""
344
345	@authenticate(self.authenticator)	2✔
346	@only_authenticated	2✔
347	@validate(json=apispec.SessionSecretSlotPost)	2✔
348	async def _post_session_secret_slot(	2✔
349	_: Request, user: base_models.APIUser, body: apispec.SessionSecretSlotPost
350	) -> JSONResponse:
351	unsaved_secret_slot = validate_unsaved_session_secret_slot(body)	2✔
352	secret_slot = await self.session_secret_repo.insert_session_secret_slot(	2✔
353	user=user, secret_slot=unsaved_secret_slot
354	)
355	return validated_json(apispec.SessionSecretSlot, secret_slot, status=201)	1✔
356
357	return "/session_secret_slots", ["POST"], _post_session_secret_slot	2✔
358
359	def get_session_secret_slot(self) -> BlueprintFactoryResponse:	2✔
360	"""Get the details of a session secret slot."""
361
362	@authenticate(self.authenticator)	2✔
363	@extract_if_none_match	2✔
364	async def _get_session_secret_slot(	2✔
365	_: Request, user: base_models.APIUser, slot_id: ULID, etag: str \| None
366	) -> HTTPResponse:
367	secret_slot = await self.session_secret_repo.get_session_secret_slot(user=user, slot_id=slot_id)	2✔
368
369	if secret_slot.etag == etag:	1✔
NEW 370	return HTTPResponse(status=304)	×
371
372	return validated_json(apispec.SessionSecretSlot, secret_slot)	1✔
373
374	return "/session_secret_slots/<slot_id:ulid>", ["GET"], _get_session_secret_slot	2✔
375
376	def patch_session_secret_slot(self) -> BlueprintFactoryResponse:	2✔
377	"""Update specific fields of an existing session secret slot."""
378
379	@authenticate(self.authenticator)	2✔
380	@only_authenticated	2✔
381	@if_match_required	2✔
382	@validate(json=apispec.SessionSecretSlotPatch)	2✔
383	async def _patch_session_secret_slot(	2✔
384	_: Request,
385	user: base_models.APIUser,
386	slot_id: ULID,
387	body: apispec.SessionSecretSlotPatch,
388	etag: str,
389	) -> JSONResponse:
390	secret_slot_patch = validate_session_secret_slot_patch(body)	2✔
391	secret_slot = await self.session_secret_repo.update_session_secret_slot(	2✔
392	user=user, slot_id=slot_id, patch=secret_slot_patch, etag=etag
393	)
394	return validated_json(apispec.SessionSecretSlot, secret_slot)	1✔
395
396	return "/session_secret_slots/<slot_id:ulid>", ["PATCH"], _patch_session_secret_slot	2✔
397
398	def delete_session_secret_slot(self) -> BlueprintFactoryResponse:	2✔
399	"""Remove a session secret slot."""
400
401	@authenticate(self.authenticator)	2✔
402	@only_authenticated	2✔
403	async def _delete_session_secret_slot(_: Request, user: base_models.APIUser, slot_id: ULID) -> HTTPResponse:	2✔
404	await self.session_secret_repo.delete_session_secret_slot(user=user, slot_id=slot_id)	2✔
405	return HTTPResponse(status=204)	2✔
406
407	return "/session_secret_slots/<slot_id:ulid>", ["DELETE"], _delete_session_secret_slot	2✔

SwissDataScienceCenter / renku-data-services / 12047437567

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