5529030370

Build Type

push

github-actions

Committed by Ralf Grubenmann

Commit Message

fix docker build, pin versions

Run Details

24252 of 28479 relevant lines covered (85.16%)

2.95 hits per line

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

29.79

/renku/core/dataset/request_model.py

# -*- coding: utf-8 -*-
#
# Copyright 2017-2022 - Swiss Data Science Center (SDSC)
# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Renku management dataset request models."""

import imghdr
import os
import shutil
import urllib
from pathlib import Path
from typing import List, Optional, Union, cast
from urllib.request import urlretrieve

from renku.core import errors
from renku.domain_model.dataset import Dataset, ImageObject
from renku.domain_model.project_context import project_context


class ImageRequestModel:
    """Model for passing image information to dataset use-cases."""

    def __init__(
        self,
        content_url: str,
        position: int,
        mirror_locally: bool = False,
        safe_image_paths: Optional[List[str]] = None,
    ) -> None:
        self.content_url = content_url
        self.position = position
        self.mirror_locally = mirror_locally
        self.safe_image_paths: List[Union[str, Path]] = cast(List[Union[str, Path]], safe_image_paths) or []

    def to_image_object(self, dataset: Dataset) -> ImageObject:
        """Convert request model to ``ImageObject``."""
        image_type = None
        self.safe_image_paths.append(project_context.path)

        image_folder = project_context.dataset_images_path / dataset.initial_identifier
        image_folder.mkdir(exist_ok=True, parents=True)

        if urllib.parse.urlparse(self.content_url).netloc:
            # NOTE: absolute url
            if not self.mirror_locally:
                return ImageObject(
                    content_url=self.content_url,
                    position=self.position,
                    id=ImageObject.generate_id(dataset_id=dataset.id, position=self.position),
                )

            # NOTE: mirror the image locally
            try:
                path, _ = urlretrieve(self.content_url)
            except urllib.error.URLError as e:
                raise errors.DatasetImageError(f"Dataset image with url {self.content_url} couldn't be mirrored") from e

            image_type = imghdr.what(path)
            if image_type:
                image_type = f".{image_type}"

            self.content_url = path
            self.safe_image_paths.append(Path(path).parent)

        path = self.content_url
        if not os.path.isabs(path):
            path = os.path.normpath(os.path.join(project_context.path, path))

        if not os.path.exists(path) or not any(
            os.path.commonprefix([path, p]) == str(p) for p in self.safe_image_paths
        ):
            # NOTE: make sure files exists and prevent path traversal
            raise errors.DatasetImageError(f"Dataset image with relative path {self.content_url} not found")

        if not path.startswith(str(image_folder)):
            # NOTE: only copy dataset image if it's not in .renku/datasets/<id>/images/ already
            if image_type:
                ext = image_type
            else:
                _, ext = os.path.splitext(self.content_url)

            img_path = image_folder / f"{self.position}{ext}"
            shutil.copy(path, img_path)
        else:
            img_path = Path(path)

        return ImageObject(
            content_url=str(img_path.relative_to(project_context.path)),
            position=self.position,
            id=ImageObject.generate_id(dataset_id=dataset.id, position=self.position),
        )

1	# -- coding: utf-8 --
2	#
3	# Copyright 2017-2022 - Swiss Data Science Center (SDSC)
4	# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
5	# Eidgenössische Technische Hochschule Zürich (ETHZ).
6	#
7	# Licensed under the Apache License, Version 2.0 (the "License");
8	# you may not use this file except in compliance with the License.
9	# You may obtain a copy of the License at
10	#
11	# http://www.apache.org/licenses/LICENSE-2.0
12	#
13	# Unless required by applicable law or agreed to in writing, software
14	# distributed under the License is distributed on an "AS IS" BASIS,
15	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16	# See the License for the specific language governing permissions and
17	# limitations under the License.
18	"""Renku management dataset request models."""	6✔
19
20	import imghdr	6✔
21	import os	6✔
22	import shutil	6✔
23	import urllib	6✔
24	from pathlib import Path	6✔
25	from typing import List, Optional, Union, cast	6✔
26	from urllib.request import urlretrieve	6✔
27
28	from renku.core import errors	6✔
29	from renku.domain_model.dataset import Dataset, ImageObject	6✔
30	from renku.domain_model.project_context import project_context	6✔
31
32
33	class ImageRequestModel:	6✔
34	"""Model for passing image information to dataset use-cases."""
35
36	def __init__(	6✔
37	self,
38	content_url: str,
39	position: int,
40	mirror_locally: bool = False,
41	safe_image_paths: Optional[List[str]] = None,
42	) -> None:
43	self.content_url = content_url	×
44	self.position = position	×
45	self.mirror_locally = mirror_locally	×
46	self.safe_image_paths: List[Union[str, Path]] = cast(List[Union[str, Path]], safe_image_paths) or []	×
47
48	def to_image_object(self, dataset: Dataset) -> ImageObject:	6✔
49	"""Convert request model to ``ImageObject``."""
50	image_type = None	×
51	self.safe_image_paths.append(project_context.path)	×
52
53	image_folder = project_context.dataset_images_path / dataset.initial_identifier	×
54	image_folder.mkdir(exist_ok=True, parents=True)	×
55
56	if urllib.parse.urlparse(self.content_url).netloc:	×
57	# NOTE: absolute url
58	if not self.mirror_locally:	×
59	return ImageObject(	×
60	content_url=self.content_url,
61	position=self.position,
62	id=ImageObject.generate_id(dataset_id=dataset.id, position=self.position),
63	)
64
65	# NOTE: mirror the image locally
66	try:	×
67	path, _ = urlretrieve(self.content_url)	×
68	except urllib.error.URLError as e:	×
69	raise errors.DatasetImageError(f"Dataset image with url {self.content_url} couldn't be mirrored") from e	×
70
71	image_type = imghdr.what(path)	×
72	if image_type:	×
73	image_type = f".{image_type}"	×
74
75	self.content_url = path	×
76	self.safe_image_paths.append(Path(path).parent)	×
77
78	path = self.content_url	×
79	if not os.path.isabs(path):	×
80	path = os.path.normpath(os.path.join(project_context.path, path))	×
81
82	if not os.path.exists(path) or not any(	×
83	os.path.commonprefix([path, p]) == str(p) for p in self.safe_image_paths
84	):
85	# NOTE: make sure files exists and prevent path traversal
86	raise errors.DatasetImageError(f"Dataset image with relative path {self.content_url} not found")	×
87
88	if not path.startswith(str(image_folder)):	×
89	# NOTE: only copy dataset image if it's not in .renku/datasets/<id>/images/ already
90	if image_type:	×
91	ext = image_type	×
92	else:
93	_, ext = os.path.splitext(self.content_url)	×
94
95	img_path = image_folder / f"{self.position}{ext}"	×
96	shutil.copy(path, img_path)	×
97	else:
98	img_path = Path(path)	×
99
100	return ImageObject(	×
101	content_url=str(img_path.relative_to(project_context.path)),
102	position=self.position,
103	id=ImageObject.generate_id(dataset_id=dataset.id, position=self.position),
104	)

SwissDataScienceCenter / renku-python / 5529030370

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