14382014257

Committed 10 Apr 2025 01:42PM UTC coverage: 86.576% (+0.2%) from 86.351%

Build # 14382014257

Build Type

Pull #759

github

Committed by

web-flow

Commit Message

Merge 470ff1568 into 74eb7d965

Pull Request Pull Request #759: feat: add new service cache and migrations

Run Details

412 of 486 new or added lines in 15 files covered. (84.77%)

18 existing lines in 6 files now uncovered.

20232 of 23369 relevant lines covered (86.58%)

1.53 hits per line

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

95.69

/components/renku_data_services/storage/blueprints.py

"""Cloud storage app."""

from dataclasses import dataclass
from typing import Any

from sanic import HTTPResponse, Request, empty
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.base_models.validation import validated_json
from renku_data_services.storage import apispec, models
from renku_data_services.storage.db import StorageRepository
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)


@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 = await self.storage_repo.get_storage(user=user, project_id=query.project_id)

            return validated_json(
                apispec.StorageGetResponse, [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 validated_json(apispec.CloudStorageGet, 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 validated_json(apispec.CloudStorageGet, 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 validated_json(apispec.CloudStorageGet, 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 validated_json(apispec.CloudStorageGet, 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 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 validated_json(apispec.RCloneSchema, validator.asdict())

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

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

        @validate(json=apispec.StorageSchemaTestConnectionPostRequest)
        async def _test_connection(
            request: Request, validator: RCloneValidator, body: apispec.StorageSchemaTestConnectionPostRequest
        ) -> HTTPResponse:
            validator.validate(body.configuration, keep_sensitive=True)
            result = await validator.test_connection(body.configuration, body.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."""

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

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

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

        @validate(json=apispec.StorageSchemaObscurePostRequest)
        async def _obscure(
            request: Request, validator: RCloneValidator, body: apispec.StorageSchemaObscurePostRequest
        ) -> JSONResponse:
            config = await validator.obscure_config(body.configuration)
            return validated_json(apispec.RCloneConfigValidate, config)

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

1	"""Cloud storage app."""
2
3	from dataclasses import dataclass	2✔
4	from typing import Any	2✔
5
6	from sanic import HTTPResponse, Request, empty	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.base_models.validation import validated_json	2✔
17	from renku_data_services.storage import apispec, models	2✔
18	from renku_data_services.storage.db import StorageRepository	2✔
19	from renku_data_services.storage.rclone import RCloneValidator	2✔
20
21
22	def dump_storage_with_sensitive_fields(storage: models.CloudStorage, validator: RCloneValidator) -> dict[str, Any]:	2✔
23	"""Dump a CloudStorage model alongside sensitive fields."""
24	return apispec.CloudStorageGet.model_validate(	1✔
25	{
26	"storage": apispec.CloudStorageWithId.model_validate(storage).model_dump(exclude_none=True),
27	"sensitive_fields": validator.get_private_fields(storage.configuration),
28	}
29	).model_dump(exclude_none=True)
30
31
32	@dataclass(kw_only=True)	2✔
33	class StorageBP(CustomBlueprint):	2✔
34	"""Handlers for manipulating storage definitions."""
35
36	storage_repo: StorageRepository	2✔
37	authenticator: base_models.Authenticator	2✔
38
39	def get(self) -> BlueprintFactoryResponse:	2✔
40	"""Get cloud storage for a repository."""
41
42	@authenticate(self.authenticator)	2✔
43	@validate_query(query=apispec.StorageParams)	2✔
44	async def _get(	2✔
45	request: Request,
46	user: base_models.APIUser,
47	validator: RCloneValidator,
48	query: apispec.StorageParams,
49	) -> JSONResponse:
50	storage = await self.storage_repo.get_storage(user=user, project_id=query.project_id)	2✔
51
52	return validated_json(	2✔
53	apispec.StorageGetResponse, [dump_storage_with_sensitive_fields(s, validator) for s in storage]
54	)
55
56	return "/storage", ["GET"], _get	2✔
57
58	def get_one(self) -> BlueprintFactoryResponse:	2✔
59	"""Get a single storage by id."""
60
61	@authenticate(self.authenticator)	2✔
62	async def _get_one(	2✔
63	request: Request,
64	user: base_models.APIUser,
65	storage_id: ULID,
66	validator: RCloneValidator,
67	) -> JSONResponse:
68	storage = await self.storage_repo.get_storage_by_id(storage_id, user=user)	2✔
69
70	return validated_json(apispec.CloudStorageGet, dump_storage_with_sensitive_fields(storage, validator))	×
71
72	return "/storage/<storage_id:ulid>", ["GET"], _get_one	2✔
73
74	def post(self) -> BlueprintFactoryResponse:	2✔
75	"""Create a new cloud storage entry."""
76
77	@authenticate(self.authenticator)	2✔
78	async def _post(request: Request, user: base_models.APIUser, validator: RCloneValidator) -> JSONResponse:	2✔
79	storage: models.UnsavedCloudStorage
80	if not isinstance(request.json, dict):	2✔
81	body_type = type(request.json)	1✔
82	raise errors.ValidationError(	1✔
83	message=f"The payload is supposed to be a dictionary, got {body_type.__name__}"
84	)
85	if "storage_url" in request.json:	2✔
86	url_body = apispec.CloudStorageUrl(**request.json)	1✔
87	storage = models.UnsavedCloudStorage.from_url(	1✔
88	storage_url=url_body.storage_url,
89	project_id=url_body.project_id.root,
90	name=url_body.name,
91	target_path=url_body.target_path,
92	readonly=url_body.readonly,
93	)
94	else:
95	body = apispec.CloudStorage(**request.json)	2✔
96	storage = models.UnsavedCloudStorage.from_dict(body.model_dump())	2✔
97
98	validator.validate(storage.configuration.model_dump())	1✔
99
100	res = await self.storage_repo.insert_storage(storage=storage, user=user)	1✔
101	return validated_json(apispec.CloudStorageGet, dump_storage_with_sensitive_fields(res, validator), 201)	1✔
102
103	return "/storage", ["POST"], _post	2✔
104
105	def put(self) -> BlueprintFactoryResponse:	2✔
106	"""Replace a storage entry."""
107
108	@authenticate(self.authenticator)	2✔
109	async def _put(	2✔
110	request: Request,
111	user: base_models.APIUser,
112	storage_id: ULID,
113	validator: RCloneValidator,
114	) -> JSONResponse:
115	if not request.json:	2✔
116	raise errors.ValidationError(message="The request body is empty. Please provide a valid JSON object.")	×
117	if not isinstance(request.json, dict):	2✔
118	raise errors.ValidationError(message="The request body is not a valid JSON object.")	×
119	if "storage_url" in request.json:	2✔
UNCOV 120	url_body = apispec.CloudStorageUrl(**request.json)	×
UNCOV 121	new_storage = models.UnsavedCloudStorage.from_url(	×
122	storage_url=url_body.storage_url,
123	project_id=url_body.project_id.root,
124	name=url_body.name,
125	target_path=url_body.target_path,
126	readonly=url_body.readonly,
127	)
128	else:
129	body = apispec.CloudStorage(**request.json)	2✔
130	new_storage = models.UnsavedCloudStorage.from_dict(body.model_dump())	2✔
131
132	validator.validate(new_storage.configuration.model_dump())	1✔
133	body_dict = new_storage.model_dump()	1✔
134	res = await self.storage_repo.update_storage(storage_id=storage_id, user=user, **body_dict)	1✔
135	return validated_json(apispec.CloudStorageGet, dump_storage_with_sensitive_fields(res, validator))	1✔
136
137	return "/storage/<storage_id:ulid>", ["PUT"], _put	2✔
138
139	def patch(self) -> BlueprintFactoryResponse:	2✔
140	"""Update parts of a storage entry."""
141
142	@authenticate(self.authenticator)	2✔
143	@validate(json=apispec.CloudStoragePatch)	2✔
144	async def _patch(	2✔
145	request: Request,
146	user: base_models.APIUser,
147	storage_id: ULID,
148	body: apispec.CloudStoragePatch,
149	validator: RCloneValidator,
150	) -> JSONResponse:
151	existing_storage = await self.storage_repo.get_storage_by_id(storage_id, user=user)	2✔
152	if body.configuration is not None:	1✔
153	# we need to apply the patch to the existing storage to properly validate it
154	body.configuration = {existing_storage.configuration, body.configuration}	1✔
155
156	for k, v in list(body.configuration.items()):	1✔
157	if v is None:	1✔
158	# delete fields that were unset
159	del body.configuration[k]	1✔
160	validator.validate(body.configuration)	1✔
161
162	body_dict = body.model_dump(exclude_none=True)	1✔
163
164	res = await self.storage_repo.update_storage(storage_id=storage_id, user=user, **body_dict)	1✔
165	return validated_json(apispec.CloudStorageGet, dump_storage_with_sensitive_fields(res, validator))	1✔
166
167	return "/storage/<storage_id:ulid>", ["PATCH"], _patch	2✔
168
169	def delete(self) -> BlueprintFactoryResponse:	2✔
170	"""Delete a storage entry."""
171
172	@authenticate(self.authenticator)	2✔
173	async def _delete(request: Request, user: base_models.APIUser, storage_id: ULID) -> HTTPResponse:	2✔
174	await self.storage_repo.delete_storage(storage_id=storage_id, user=user)	2✔
175	return empty(204)	2✔
176
177	return "/storage/<storage_id:ulid>", ["DELETE"], _delete	2✔
178
179
180	@dataclass(kw_only=True)	2✔
181	class StorageSchemaBP(CustomBlueprint):	2✔
182	"""Handler for getting RClone storage schema."""
183
184	def get(self) -> BlueprintFactoryResponse:	2✔
185	"""Get cloud storage for a repository."""
186
187	async def _get(_: Request, validator: RCloneValidator) -> JSONResponse:	2✔
188	return validated_json(apispec.RCloneSchema, validator.asdict())	2✔
189
190	return "/storage_schema", ["GET"], _get	2✔
191
192	def test_connection(self) -> BlueprintFactoryResponse:	2✔
193	"""Validate an RClone config."""
194
195	@validate(json=apispec.StorageSchemaTestConnectionPostRequest)	2✔
196	async def _test_connection(	2✔
197	request: Request, validator: RCloneValidator, body: apispec.StorageSchemaTestConnectionPostRequest
198	) -> HTTPResponse:
199	validator.validate(body.configuration, keep_sensitive=True)	2✔
200	result = await validator.test_connection(body.configuration, body.source_path)	1✔
201	if not result.success:	1✔
202	raise errors.ValidationError(message=result.error)	1✔
203	return empty(204)	1✔
204
205	return "/storage_schema/test_connection", ["POST"], _test_connection	2✔
206
207	def validate(self) -> BlueprintFactoryResponse:	2✔
208	"""Validate an RClone config."""
209
210	@validate(json=apispec.RCloneConfigValidate)	2✔
211	async def _validate(	2✔
212	request: Request, validator: RCloneValidator, body: apispec.RCloneConfigValidate
213	) -> HTTPResponse:
214	if body.root is None:	2✔
215	raise errors.ValidationError(message="The request body is empty. Please provide a valid JSON object.")	2✔
216	validator.validate(body.root, keep_sensitive=True)	2✔
217	return empty(204)	1✔
218
219	return "/storage_schema/validate", ["POST"], _validate	2✔
220
221	def obscure(self) -> BlueprintFactoryResponse:	2✔
222	"""Obscure values in config."""
223
224	@validate(json=apispec.StorageSchemaObscurePostRequest)	2✔
225	async def _obscure(	2✔
226	request: Request, validator: RCloneValidator, body: apispec.StorageSchemaObscurePostRequest
227	) -> JSONResponse:
228	config = await validator.obscure_config(body.configuration)	2✔
229	return validated_json(apispec.RCloneConfigValidate, config)	1✔
230
231	return "/storage_schema/obscure", ["POST"], _obscure	2✔

SwissDataScienceCenter / renku-data-services / 14382014257

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