11034384359

Committed 25 Sep 2024 01:46PM UTC coverage: 90.429% (+0.03%) from 90.403%

Build # 11034384359

Build Type

Pull #425

github

Committed by

web-flow

Commit Message

Merge 871b24a54 into 013683161

Pull Request Pull Request #425: refactor: define a single method to remove entities from authz

Run Details

28 of 33 new or added lines in 1 file covered. (84.85%)

2 existing lines in 2 files now uncovered.

9335 of 10323 relevant lines covered (90.43%)

1.6 hits per line

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

95.24

/components/renku_data_services/storage/blueprints.py

"""Cloud storage app."""

from dataclasses import dataclass
from typing import Any

from sanic import HTTPResponse, Request, empty, json
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 import errors
from renku_data_services.base_api.auth import authenticate
from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint
from renku_data_services.base_api.misc import validate_query
from renku_data_services.storage import apispec, models
from renku_data_services.storage.db import StorageRepository, StorageV2Repository
from renku_data_services.storage.rclone import RCloneValidator


def dump_storage_with_sensitive_fields(storage: models.CloudStorage, validator: RCloneValidator) -> dict[str, Any]:
    """Dump a CloudStorage model alongside sensitive fields."""
    return apispec.CloudStorageGet.model_validate(
        {
            "storage": apispec.CloudStorageWithId.model_validate(storage).model_dump(exclude_none=True),
            "sensitive_fields": validator.get_private_fields(storage.configuration),
        }
    ).model_dump(exclude_none=True)


def dump_storage_with_sensitive_fields_and_secrets(
    storage: models.CloudStorage, validator: RCloneValidator
) -> dict[str, Any]:
    """Dump a CloudStorage model alongside sensitive fields and its saved secrets."""
    dumped_storage = dump_storage_with_sensitive_fields(storage, validator)
    dumped_storage["secrets"] = [apispec.CloudStorageSecretGet.model_validate(s).model_dump() for s in storage.secrets]
    return dumped_storage


@dataclass(kw_only=True)
class StorageBP(CustomBlueprint):
    """Handlers for manipulating storage definitions."""

    storage_repo: StorageRepository
    authenticator: base_models.Authenticator

    def get(self) -> BlueprintFactoryResponse:
        """Get cloud storage for a repository."""

        @authenticate(self.authenticator)
        @validate_query(query=apispec.StorageParams)
        async def _get(
            request: Request,
            user: base_models.APIUser,
            validator: RCloneValidator,
            query: apispec.StorageParams,
        ) -> JSONResponse:
            storage: list[models.CloudStorage]
            storage = await self.storage_repo.get_storage(user=user, project_id=query.project_id)

            return json([dump_storage_with_sensitive_fields(s, validator) for s in storage])

        return "/storage", ["GET"], _get

    def get_one(self) -> BlueprintFactoryResponse:
        """Get a single storage by id."""

        @authenticate(self.authenticator)
        async def _get_one(
            request: Request,
            user: base_models.APIUser,
            storage_id: ULID,
            validator: RCloneValidator,
        ) -> JSONResponse:
            storage = await self.storage_repo.get_storage_by_id(storage_id, user=user)

            return json(dump_storage_with_sensitive_fields(storage, validator))

        return "/storage/<storage_id:ulid>", ["GET"], _get_one

    def post(self) -> BlueprintFactoryResponse:
        """Create a new cloud storage entry."""

        @authenticate(self.authenticator)
        async def _post(request: Request, user: base_models.APIUser, validator: RCloneValidator) -> JSONResponse:
            storage: models.UnsavedCloudStorage
            if not isinstance(request.json, dict):
                body_type = type(request.json)
                raise errors.ValidationError(
                    message=f"The payload is supposed to be a dictionary, got {body_type.__name__}"
                )
            if "storage_url" in request.json:
                url_body = apispec.CloudStorageUrl(**request.json)
                storage = models.UnsavedCloudStorage.from_url(
                    storage_url=url_body.storage_url,
                    project_id=url_body.project_id.root,
                    name=url_body.name,
                    target_path=url_body.target_path,
                    readonly=url_body.readonly,
                )
            else:
                body = apispec.CloudStorage(**request.json)
                storage = models.UnsavedCloudStorage.from_dict(body.model_dump())

            validator.validate(storage.configuration.model_dump())

            res = await self.storage_repo.insert_storage(storage=storage, user=user)
            return json(dump_storage_with_sensitive_fields(res, validator), 201)

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

    def put(self) -> BlueprintFactoryResponse:
        """Replace a storage entry."""

        @authenticate(self.authenticator)
        async def _put(
            request: Request,
            user: base_models.APIUser,
            storage_id: ULID,
            validator: RCloneValidator,
        ) -> JSONResponse:
            if not request.json:
                raise errors.ValidationError(message="The request body is empty. Please provide a valid JSON object.")
            if not isinstance(request.json, dict):
                raise errors.ValidationError(message="The request body is not a valid JSON object.")
            if "storage_url" in request.json:
                url_body = apispec.CloudStorageUrl(**request.json)
                new_storage = models.UnsavedCloudStorage.from_url(
                    storage_url=url_body.storage_url,
                    project_id=url_body.project_id.root,
                    name=url_body.name,
                    target_path=url_body.target_path,
                    readonly=url_body.readonly,
                )
            else:
                body = apispec.CloudStorage(**request.json)
                new_storage = models.UnsavedCloudStorage.from_dict(body.model_dump())

            validator.validate(new_storage.configuration.model_dump())
            body_dict = new_storage.model_dump()
            res = await self.storage_repo.update_storage(storage_id=storage_id, user=user, **body_dict)
            return json(dump_storage_with_sensitive_fields(res, validator))

        return "/storage/<storage_id:ulid>", ["PUT"], _put

    def patch(self) -> BlueprintFactoryResponse:
        """Update parts of a storage entry."""

        @authenticate(self.authenticator)
        @validate(json=apispec.CloudStoragePatch)
        async def _patch(
            request: Request,
            user: base_models.APIUser,
            storage_id: ULID,
            body: apispec.CloudStoragePatch,
            validator: RCloneValidator,
        ) -> JSONResponse:
            existing_storage = await self.storage_repo.get_storage_by_id(storage_id, user=user)
            if body.configuration is not None:
                # we need to apply the patch to the existing storage to properly validate it
                body.configuration = {**existing_storage.configuration, **body.configuration}

                for k, v in list(body.configuration.items()):
                    if v is None:
                        # delete fields that were unset
                        del body.configuration[k]
                validator.validate(body.configuration)

            body_dict = body.model_dump(exclude_none=True)

            res = await self.storage_repo.update_storage(storage_id=storage_id, user=user, **body_dict)
            return json(dump_storage_with_sensitive_fields(res, validator))

        return "/storage/<storage_id:ulid>", ["PATCH"], _patch

    def delete(self) -> BlueprintFactoryResponse:
        """Delete a storage entry."""

        @authenticate(self.authenticator)
        async def _delete(request: Request, user: base_models.APIUser, storage_id: ULID) -> HTTPResponse:
            await self.storage_repo.delete_storage(storage_id=storage_id, user=user)
            return empty(204)

        return "/storage/<storage_id:ulid>", ["DELETE"], _delete


@dataclass(kw_only=True)
class StoragesV2BP(CustomBlueprint):
    """Handlers for manipulating storage definitions."""

    storage_v2_repo: StorageV2Repository
    authenticator: base_models.Authenticator

    def get(self) -> BlueprintFactoryResponse:
        """Get cloud storage for a repository."""

        @authenticate(self.authenticator)
        @validate_query(query=apispec.StorageV2Params)
        async def _get(
            request: Request,
            user: base_models.APIUser,
            validator: RCloneValidator,
            query: apispec.StorageV2Params,
        ) -> JSONResponse:
            storage: list[models.CloudStorage]
            storage = await self.storage_v2_repo.get_storage(
                user=user, include_secrets=True, project_id=query.project_id
            )

            return json([dump_storage_with_sensitive_fields_and_secrets(s, validator) for s in storage])

        return "/storages_v2", ["GET"], _get

    def get_one(self) -> BlueprintFactoryResponse:
        """Get a single storage by id."""

        @authenticate(self.authenticator)
        async def _get_one(
            request: Request,
            user: base_models.APIUser,
            storage_id: ULID,
            validator: RCloneValidator,
        ) -> JSONResponse:
            storage = await self.storage_v2_repo.get_storage_by_id(storage_id, user=user)

            return json(dump_storage_with_sensitive_fields_and_secrets(storage, validator))

        return "/storages_v2/<storage_id:ulid>", ["GET"], _get_one

    def post(self) -> BlueprintFactoryResponse:
        """Create a new cloud storage entry."""

        @authenticate(self.authenticator)
        async def _post(request: Request, user: base_models.APIUser, validator: RCloneValidator) -> JSONResponse:
            storage: models.UnsavedCloudStorage
            if not isinstance(request.json, dict):
                body_type = type(request.json)
                raise errors.ValidationError(
                    message=f"The payload is supposed to be a dictionary, got {body_type.__name__}"
                )
            if "storage_url" in request.json:
                url_body = apispec.CloudStorageUrl(**request.json)
                storage = models.UnsavedCloudStorage.from_url(
                    storage_url=url_body.storage_url,
                    project_id=url_body.project_id.root,
                    name=url_body.name,
                    target_path=url_body.target_path,
                    readonly=url_body.readonly,
                )
            else:
                body = apispec.CloudStorage(**request.json)
                storage = models.UnsavedCloudStorage.from_dict(body.model_dump())

            validator.validate(storage.configuration.model_dump())

            res = await self.storage_v2_repo.insert_storage(storage=storage, user=user)
            return json(dump_storage_with_sensitive_fields(res, validator), 201)

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

    def patch(self) -> BlueprintFactoryResponse:
        """Update parts of a storage entry."""

        @authenticate(self.authenticator)
        @validate(json=apispec.CloudStoragePatch)
        async def _patch(
            _: Request,
            user: base_models.APIUser,
            storage_id: ULID,
            body: apispec.CloudStoragePatch,
            validator: RCloneValidator,
        ) -> JSONResponse:
            existing_storage = await self.storage_v2_repo.get_storage_by_id(storage_id, user=user)
            if body.configuration is not None:
                # we need to apply the patch to the existing storage to properly validate it
                body.configuration = {**existing_storage.configuration, **body.configuration}

                for k, v in list(body.configuration.items()):
                    if v is None:
                        # delete fields that were unset
                        del body.configuration[k]
                validator.validate(body.configuration)

            body_dict = body.model_dump(exclude_none=True)

            res = await self.storage_v2_repo.update_storage(storage_id=storage_id, user=user, **body_dict)
            return json(dump_storage_with_sensitive_fields(res, validator))

        return "/storages_v2/<storage_id:ulid>", ["PATCH"], _patch

    def delete(self) -> BlueprintFactoryResponse:
        """Delete a storage entry."""

        @authenticate(self.authenticator)
        async def _delete(request: Request, user: base_models.APIUser, storage_id: ULID) -> HTTPResponse:
            await self.storage_v2_repo.delete_storage(storage_id=storage_id, user=user)
            return empty(204)

        return "/storages_v2/<storage_id:ulid>", ["DELETE"], _delete

    def upsert_secrets(self) -> BlueprintFactoryResponse:
        """Create/update secrets for a cloud storage."""

        @authenticate(self.authenticator)
        async def _upsert_secrets(request: Request, user: base_models.APIUser, storage_id: ULID) -> JSONResponse:
            # TODO: use @validate once sanic supports validating json lists
            body = apispec.CloudStorageSecretPostList.model_validate(request.json)
            secrets = [models.CloudStorageSecretUpsert.model_validate(s.model_dump()) for s in body.root]
            result = await self.storage_v2_repo.upsert_storage_secrets(
                storage_id=storage_id, user=user, secrets=secrets
            )
            return json(
                apispec.CloudStorageSecretGetList.model_validate(result).model_dump(exclude_none=True, mode="json"), 201
            )

        return "/storages_v2/<storage_id:ulid>/secrets", ["POST"], _upsert_secrets

    def get_secrets(self) -> BlueprintFactoryResponse:
        """Return all secrets for a cloud storage."""

        @authenticate(self.authenticator)
        async def _get_secrets(request: Request, user: base_models.APIUser, storage_id: ULID) -> JSONResponse:
            result = await self.storage_v2_repo.get_storage_secrets(storage_id=storage_id, user=user)
            return json(
                apispec.CloudStorageSecretGetList.model_validate(result).model_dump(exclude_none=True, mode="json"), 200
            )

        return "/storages_v2/<storage_id:ulid>/secrets", ["GET"], _get_secrets

    def delete_secrets(self) -> BlueprintFactoryResponse:
        """Delete all secrets for a cloud storage."""

        @authenticate(self.authenticator)
        async def _delete_secrets(request: Request, user: base_models.APIUser, storage_id: ULID) -> HTTPResponse:
            await self.storage_v2_repo.delete_storage_secrets(storage_id=storage_id, user=user)
            return HTTPResponse(status=204)

        return "/storages_v2/<storage_id:ulid>/secrets", ["DELETE"], _delete_secrets


@dataclass(kw_only=True)
class StorageSchemaBP(CustomBlueprint):
    """Handler for getting RClone storage schema."""

    def get(self) -> BlueprintFactoryResponse:
        """Get cloud storage for a repository."""

        async def _get(_: Request, validator: RCloneValidator) -> JSONResponse:
            return json(validator.asdict())

        return "/storage_schema", ["GET"], _get

    def test_connection(self) -> BlueprintFactoryResponse:
        """Validate an RClone config."""

        async def _test_connection(request: Request, validator: RCloneValidator) -> HTTPResponse:
            if not request.json:
                raise errors.ValidationError(message="The request body is empty. Please provide a valid JSON object.")
            if not isinstance(request.json, dict):
                raise errors.ValidationError(message="The request body is not a valid JSON object.")
            if not request.json.get("configuration"):
                raise errors.ValidationError(message="No 'configuration' sent.")
            if not isinstance(request.json.get("configuration"), dict):
                config_type = type(request.json.get("configuration"))
                raise errors.ValidationError(
                    message=f"The R clone configuration should be a dictionary, not {config_type.__name__}"
                )
            if not request.json.get("source_path"):
                raise errors.ValidationError(message="'source_path' is required to test the connection.")
            validator.validate(request.json["configuration"], keep_sensitive=True)
            result = await validator.test_connection(request.json["configuration"], request.json["source_path"])
            if not result.success:
                raise errors.ValidationError(message=result.error)
            return empty(204)

        return "/storage_schema/test_connection", ["POST"], _test_connection

    def validate(self) -> BlueprintFactoryResponse:
        """Validate an RClone config."""

        async def _validate(request: Request, validator: RCloneValidator) -> HTTPResponse:
            if not request.json:
                raise errors.ValidationError(message="The request body is empty. Please provide a valid JSON object.")
            if not isinstance(request.json, dict):
                raise errors.ValidationError(message="The request body is not a valid JSON object.")
            validator.validate(request.json, keep_sensitive=True)
            return empty(204)

        return "/storage_schema/validate", ["POST"], _validate

    def obscure(self) -> BlueprintFactoryResponse:
        """Obscure values in config."""

        async def _obscure(request: Request, validator: RCloneValidator) -> JSONResponse:
            if not request.json:
                raise errors.ValidationError(message="The request body is empty. Please provide a valid JSON object.")
            if not isinstance(request.json, dict):
                raise errors.ValidationError(message="The request body is not a valid JSON object.")
            config = await validator.obscure_config(request.json)
            return json(config)

        return "/storage_schema/obscure", ["POST"], _obscure

1	"""Cloud storage app."""	2✔
2
3	from dataclasses import dataclass	2✔
4	from typing import Any	2✔
5
6	from sanic import HTTPResponse, Request, empty, json	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 import errors	2✔
13	from renku_data_services.base_api.auth import authenticate	2✔
14	from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint	2✔
15	from renku_data_services.base_api.misc import validate_query	2✔
16	from renku_data_services.storage import apispec, models	2✔
17	from renku_data_services.storage.db import StorageRepository, StorageV2Repository	2✔
18	from renku_data_services.storage.rclone import RCloneValidator	2✔
19
20
21	def dump_storage_with_sensitive_fields(storage: models.CloudStorage, validator: RCloneValidator) -> dict[str, Any]:	2✔
22	"""Dump a CloudStorage model alongside sensitive fields."""
23	return apispec.CloudStorageGet.model_validate(	1✔
24	{
25	"storage": apispec.CloudStorageWithId.model_validate(storage).model_dump(exclude_none=True),
26	"sensitive_fields": validator.get_private_fields(storage.configuration),
27	}
28	).model_dump(exclude_none=True)
29
30
31	def dump_storage_with_sensitive_fields_and_secrets(	2✔
32	storage: models.CloudStorage, validator: RCloneValidator
33	) -> dict[str, Any]:
34	"""Dump a CloudStorage model alongside sensitive fields and its saved secrets."""
35	dumped_storage = dump_storage_with_sensitive_fields(storage, validator)	1✔
36	dumped_storage["secrets"] = [apispec.CloudStorageSecretGet.model_validate(s).model_dump() for s in storage.secrets]	1✔
37	return dumped_storage	1✔
38
39
40	@dataclass(kw_only=True)	2✔
41	class StorageBP(CustomBlueprint):	2✔
42	"""Handlers for manipulating storage definitions."""	2✔
43
44	storage_repo: StorageRepository	2✔
45	authenticator: base_models.Authenticator	2✔
46
47	def get(self) -> BlueprintFactoryResponse:	2✔
48	"""Get cloud storage for a repository."""
49
50	@authenticate(self.authenticator)	2✔
51	@validate_query(query=apispec.StorageParams)	2✔
52	async def _get(	2✔
53	request: Request,
54	user: base_models.APIUser,
55	validator: RCloneValidator,
56	query: apispec.StorageParams,
57	) -> JSONResponse:
58	storage: list[models.CloudStorage]
59	storage = await self.storage_repo.get_storage(user=user, project_id=query.project_id)	2✔
60
61	return json([dump_storage_with_sensitive_fields(s, validator) for s in storage])	2✔
62
63	return "/storage", ["GET"], _get	2✔
64
65	def get_one(self) -> BlueprintFactoryResponse:	2✔
66	"""Get a single storage by id."""
67
68	@authenticate(self.authenticator)	2✔
69	async def _get_one(	2✔
70	request: Request,
71	user: base_models.APIUser,
72	storage_id: ULID,
73	validator: RCloneValidator,
74	) -> JSONResponse:
75	storage = await self.storage_repo.get_storage_by_id(storage_id, user=user)	1✔
76
77	return json(dump_storage_with_sensitive_fields(storage, validator))	×
78
79	return "/storage/<storage_id:ulid>", ["GET"], _get_one	2✔
80
81	def post(self) -> BlueprintFactoryResponse:	2✔
82	"""Create a new cloud storage entry."""
83
84	@authenticate(self.authenticator)	2✔
85	async def _post(request: Request, user: base_models.APIUser, validator: RCloneValidator) -> JSONResponse:	2✔
86	storage: models.UnsavedCloudStorage
87	if not isinstance(request.json, dict):	2✔
88	body_type = type(request.json)	1✔
89	raise errors.ValidationError(	1✔
90	message=f"The payload is supposed to be a dictionary, got {body_type.__name__}"
91	)
92	if "storage_url" in request.json:	1✔
93	url_body = apispec.CloudStorageUrl(**request.json)	1✔
94	storage = models.UnsavedCloudStorage.from_url(	1✔
95	storage_url=url_body.storage_url,
96	project_id=url_body.project_id.root,
97	name=url_body.name,
98	target_path=url_body.target_path,
99	readonly=url_body.readonly,
100	)
101	else:
102	body = apispec.CloudStorage(**request.json)	1✔
103	storage = models.UnsavedCloudStorage.from_dict(body.model_dump())	1✔
104
105	validator.validate(storage.configuration.model_dump())	1✔
106
107	res = await self.storage_repo.insert_storage(storage=storage, user=user)	1✔
108	return json(dump_storage_with_sensitive_fields(res, validator), 201)	1✔
109
110	return "/storage", ["POST"], _post	2✔
111
112	def put(self) -> BlueprintFactoryResponse:	2✔
113	"""Replace a storage entry."""
114
115	@authenticate(self.authenticator)	2✔
116	async def _put(	2✔
117	request: Request,
118	user: base_models.APIUser,
119	storage_id: ULID,
120	validator: RCloneValidator,
121	) -> JSONResponse:
122	if not request.json:	1✔
123	raise errors.ValidationError(message="The request body is empty. Please provide a valid JSON object.")	×
124	if not isinstance(request.json, dict):	1✔
125	raise errors.ValidationError(message="The request body is not a valid JSON object.")	×
126	if "storage_url" in request.json:	1✔
127	url_body = apispec.CloudStorageUrl(**request.json)	×
128	new_storage = models.UnsavedCloudStorage.from_url(	×
129	storage_url=url_body.storage_url,
130	project_id=url_body.project_id.root,
131	name=url_body.name,
132	target_path=url_body.target_path,
133	readonly=url_body.readonly,
134	)
135	else:
136	body = apispec.CloudStorage(**request.json)	1✔
137	new_storage = models.UnsavedCloudStorage.from_dict(body.model_dump())	1✔
138
139	validator.validate(new_storage.configuration.model_dump())	1✔
140	body_dict = new_storage.model_dump()	1✔
141	res = await self.storage_repo.update_storage(storage_id=storage_id, user=user, **body_dict)	1✔
142	return json(dump_storage_with_sensitive_fields(res, validator))	1✔
143
144	return "/storage/<storage_id:ulid>", ["PUT"], _put	2✔
145
146	def patch(self) -> BlueprintFactoryResponse:	2✔
147	"""Update parts of a storage entry."""
148
149	@authenticate(self.authenticator)	2✔
150	@validate(json=apispec.CloudStoragePatch)	2✔
151	async def _patch(	2✔
152	request: Request,
153	user: base_models.APIUser,
154	storage_id: ULID,
155	body: apispec.CloudStoragePatch,
156	validator: RCloneValidator,
157	) -> JSONResponse:
158	existing_storage = await self.storage_repo.get_storage_by_id(storage_id, user=user)	1✔
159	if body.configuration is not None:	1✔
160	# we need to apply the patch to the existing storage to properly validate it
161	body.configuration = {existing_storage.configuration, body.configuration}	1✔
162
163	for k, v in list(body.configuration.items()):	1✔
164	if v is None:	1✔
165	# delete fields that were unset
166	del body.configuration[k]	1✔
167	validator.validate(body.configuration)	1✔
168
169	body_dict = body.model_dump(exclude_none=True)	1✔
170
171	res = await self.storage_repo.update_storage(storage_id=storage_id, user=user, **body_dict)	1✔
172	return json(dump_storage_with_sensitive_fields(res, validator))	1✔
173
174	return "/storage/<storage_id:ulid>", ["PATCH"], _patch	2✔
175
176	def delete(self) -> BlueprintFactoryResponse:	2✔
177	"""Delete a storage entry."""
178
179	@authenticate(self.authenticator)	2✔
180	async def _delete(request: Request, user: base_models.APIUser, storage_id: ULID) -> HTTPResponse:	2✔
181	await self.storage_repo.delete_storage(storage_id=storage_id, user=user)	1✔
182	return empty(204)	1✔
183
184	return "/storage/<storage_id:ulid>", ["DELETE"], _delete	2✔
185
186
187	@dataclass(kw_only=True)	2✔
188	class StoragesV2BP(CustomBlueprint):	2✔
189	"""Handlers for manipulating storage definitions."""	2✔
190
191	storage_v2_repo: StorageV2Repository	2✔
192	authenticator: base_models.Authenticator	2✔
193
194	def get(self) -> BlueprintFactoryResponse:	2✔
195	"""Get cloud storage for a repository."""
196
197	@authenticate(self.authenticator)	2✔
198	@validate_query(query=apispec.StorageV2Params)	2✔
199	async def _get(	2✔
200	request: Request,
201	user: base_models.APIUser,
202	validator: RCloneValidator,
203	query: apispec.StorageV2Params,
204	) -> JSONResponse:
205	storage: list[models.CloudStorage]
206	storage = await self.storage_v2_repo.get_storage(	1✔
207	user=user, include_secrets=True, project_id=query.project_id
208	)
209
210	return json([dump_storage_with_sensitive_fields_and_secrets(s, validator) for s in storage])	1✔
211
212	return "/storages_v2", ["GET"], _get	2✔
213
214	def get_one(self) -> BlueprintFactoryResponse:	2✔
215	"""Get a single storage by id."""
216
217	@authenticate(self.authenticator)	2✔
218	async def _get_one(	2✔
219	request: Request,
220	user: base_models.APIUser,
221	storage_id: ULID,
222	validator: RCloneValidator,
223	) -> JSONResponse:
224	storage = await self.storage_v2_repo.get_storage_by_id(storage_id, user=user)	1✔
225
226	return json(dump_storage_with_sensitive_fields_and_secrets(storage, validator))	1✔
227
228	return "/storages_v2/<storage_id:ulid>", ["GET"], _get_one	2✔
229
230	def post(self) -> BlueprintFactoryResponse:	2✔
231	"""Create a new cloud storage entry."""
232
233	@authenticate(self.authenticator)	2✔
234	async def _post(request: Request, user: base_models.APIUser, validator: RCloneValidator) -> JSONResponse:	2✔
235	storage: models.UnsavedCloudStorage
236	if not isinstance(request.json, dict):	2✔
237	body_type = type(request.json)	1✔
238	raise errors.ValidationError(	1✔
239	message=f"The payload is supposed to be a dictionary, got {body_type.__name__}"
240	)
241	if "storage_url" in request.json:	1✔
242	url_body = apispec.CloudStorageUrl(**request.json)	×
243	storage = models.UnsavedCloudStorage.from_url(	×
244	storage_url=url_body.storage_url,
245	project_id=url_body.project_id.root,
246	name=url_body.name,
247	target_path=url_body.target_path,
248	readonly=url_body.readonly,
249	)
250	else:
251	body = apispec.CloudStorage(**request.json)	1✔
252	storage = models.UnsavedCloudStorage.from_dict(body.model_dump())	1✔
253
254	validator.validate(storage.configuration.model_dump())	1✔
255
256	res = await self.storage_v2_repo.insert_storage(storage=storage, user=user)	1✔
257	return json(dump_storage_with_sensitive_fields(res, validator), 201)	1✔
258
259	return "/storages_v2", ["POST"], _post	2✔
260
261	def patch(self) -> BlueprintFactoryResponse:	2✔
262	"""Update parts of a storage entry."""
263
264	@authenticate(self.authenticator)	2✔
265	@validate(json=apispec.CloudStoragePatch)	2✔
266	async def _patch(	2✔
267	_: Request,
268	user: base_models.APIUser,
269	storage_id: ULID,
270	body: apispec.CloudStoragePatch,
271	validator: RCloneValidator,
272	) -> JSONResponse:
273	existing_storage = await self.storage_v2_repo.get_storage_by_id(storage_id, user=user)	1✔
274	if body.configuration is not None:	1✔
275	# we need to apply the patch to the existing storage to properly validate it
276	body.configuration = {existing_storage.configuration, body.configuration}	1✔
277
278	for k, v in list(body.configuration.items()):	1✔
279	if v is None:	1✔
280	# delete fields that were unset
281	del body.configuration[k]	1✔
282	validator.validate(body.configuration)	1✔
283
284	body_dict = body.model_dump(exclude_none=True)	1✔
285
286	res = await self.storage_v2_repo.update_storage(storage_id=storage_id, user=user, **body_dict)	1✔
287	return json(dump_storage_with_sensitive_fields(res, validator))	1✔
288
289	return "/storages_v2/<storage_id:ulid>", ["PATCH"], _patch	2✔
290
291	def delete(self) -> BlueprintFactoryResponse:	2✔
292	"""Delete a storage entry."""
293
294	@authenticate(self.authenticator)	2✔
295	async def _delete(request: Request, user: base_models.APIUser, storage_id: ULID) -> HTTPResponse:	2✔
296	await self.storage_v2_repo.delete_storage(storage_id=storage_id, user=user)	1✔
297	return empty(204)	1✔
298
299	return "/storages_v2/<storage_id:ulid>", ["DELETE"], _delete	2✔
300
301	def upsert_secrets(self) -> BlueprintFactoryResponse:	2✔
302	"""Create/update secrets for a cloud storage."""
303
304	@authenticate(self.authenticator)	2✔
305	async def _upsert_secrets(request: Request, user: base_models.APIUser, storage_id: ULID) -> JSONResponse:	2✔
306	# TODO: use @validate once sanic supports validating json lists
307	body = apispec.CloudStorageSecretPostList.model_validate(request.json)	1✔
308	secrets = [models.CloudStorageSecretUpsert.model_validate(s.model_dump()) for s in body.root]	1✔
309	result = await self.storage_v2_repo.upsert_storage_secrets(	1✔
310	storage_id=storage_id, user=user, secrets=secrets
311	)
312	return json(	1✔
313	apispec.CloudStorageSecretGetList.model_validate(result).model_dump(exclude_none=True, mode="json"), 201
314	)
315
316	return "/storages_v2/<storage_id:ulid>/secrets", ["POST"], _upsert_secrets	2✔
317
318	def get_secrets(self) -> BlueprintFactoryResponse:	2✔
319	"""Return all secrets for a cloud storage."""
320
321	@authenticate(self.authenticator)	2✔
322	async def _get_secrets(request: Request, user: base_models.APIUser, storage_id: ULID) -> JSONResponse:	2✔
323	result = await self.storage_v2_repo.get_storage_secrets(storage_id=storage_id, user=user)	1✔
324	return json(	1✔
325	apispec.CloudStorageSecretGetList.model_validate(result).model_dump(exclude_none=True, mode="json"), 200
326	)
327
328	return "/storages_v2/<storage_id:ulid>/secrets", ["GET"], _get_secrets	2✔
329
330	def delete_secrets(self) -> BlueprintFactoryResponse:	2✔
331	"""Delete all secrets for a cloud storage."""
332
333	@authenticate(self.authenticator)	2✔
334	async def _delete_secrets(request: Request, user: base_models.APIUser, storage_id: ULID) -> HTTPResponse:	2✔
335	await self.storage_v2_repo.delete_storage_secrets(storage_id=storage_id, user=user)	1✔
336	return HTTPResponse(status=204)	1✔
337
338	return "/storages_v2/<storage_id:ulid>/secrets", ["DELETE"], _delete_secrets	2✔
339
340
341	@dataclass(kw_only=True)	2✔
342	class StorageSchemaBP(CustomBlueprint):	2✔
343	"""Handler for getting RClone storage schema."""	2✔
344
345	def get(self) -> BlueprintFactoryResponse:	2✔
346	"""Get cloud storage for a repository."""
347
348	async def _get(_: Request, validator: RCloneValidator) -> JSONResponse:	2✔
349	return json(validator.asdict())	2✔
350
351	return "/storage_schema", ["GET"], _get	2✔
352
353	def test_connection(self) -> BlueprintFactoryResponse:	2✔
354	"""Validate an RClone config."""
355
356	async def _test_connection(request: Request, validator: RCloneValidator) -> HTTPResponse:	2✔
357	if not request.json:	2✔
358	raise errors.ValidationError(message="The request body is empty. Please provide a valid JSON object.")	1✔
359	if not isinstance(request.json, dict):	2✔
UNCOV 360	raise errors.ValidationError(message="The request body is not a valid JSON object.")	×
361	if not request.json.get("configuration"):	2✔
362	raise errors.ValidationError(message="No 'configuration' sent.")	1✔
363	if not isinstance(request.json.get("configuration"), dict):	2✔
364	config_type = type(request.json.get("configuration"))	×
365	raise errors.ValidationError(	×
366	message=f"The R clone configuration should be a dictionary, not {config_type.__name__}"
367	)
368	if not request.json.get("source_path"):	2✔
369	raise errors.ValidationError(message="'source_path' is required to test the connection.")	2✔
370	validator.validate(request.json["configuration"], keep_sensitive=True)	1✔
371	result = await validator.test_connection(request.json["configuration"], request.json["source_path"])	1✔
372	if not result.success:	1✔
373	raise errors.ValidationError(message=result.error)	1✔
374	return empty(204)	1✔
375
376	return "/storage_schema/test_connection", ["POST"], _test_connection	2✔
377
378	def validate(self) -> BlueprintFactoryResponse:	2✔
379	"""Validate an RClone config."""
380
381	async def _validate(request: Request, validator: RCloneValidator) -> HTTPResponse:	2✔
382	if not request.json:	2✔
383	raise errors.ValidationError(message="The request body is empty. Please provide a valid JSON object.")	2✔
384	if not isinstance(request.json, dict):	2✔
385	raise errors.ValidationError(message="The request body is not a valid JSON object.")	1✔
386	validator.validate(request.json, keep_sensitive=True)	1✔
387	return empty(204)	1✔
388
389	return "/storage_schema/validate", ["POST"], _validate	2✔
390
391	def obscure(self) -> BlueprintFactoryResponse:	2✔
392	"""Obscure values in config."""
393
394	async def _obscure(request: Request, validator: RCloneValidator) -> JSONResponse:	2✔
395	if not request.json:	2✔
396	raise errors.ValidationError(message="The request body is empty. Please provide a valid JSON object.")	1✔
397	if not isinstance(request.json, dict):	2✔
398	raise errors.ValidationError(message="The request body is not a valid JSON object.")	1✔
399	config = await validator.obscure_config(request.json)	2✔
400	return json(config)	1✔
401
402	return "/storage_schema/obscure", ["POST"], _obscure	2✔

SwissDataScienceCenter / renku-data-services / 11034384359

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