14377770991

Committed 10 Apr 2025 10:04AM UTC coverage: 86.244% (-0.1%) from 86.351%

Build # 14377770991

Build Type

Pull #785

github

Committed by

web-flow

Commit Message

Merge 5b3aea87d into 74eb7d965

Pull Request Pull Request #785: feat: add endpoint to get v1 project properties

Run Details

32 of 73 new or added lines in 6 files covered. (43.84%)

1 existing line in 1 file now uncovered.

20013 of 23205 relevant lines covered (86.24%)

1.53 hits per line

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

87.34

/components/renku_data_services/base_models/core.py

"""Base models shared by services."""

from __future__ import annotations

import re
import unicodedata
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, StrEnum
from typing import ClassVar, Never, NewType, Optional, Protocol, Self, TypeVar, overload

from sanic import Request

from renku_data_services.errors import errors


@dataclass(kw_only=True, frozen=True)
class APIUser:
    """The model for a user of the API, used for authentication."""

    id: str | None = None  # the sub claim in the access token - i.e. the Keycloak user ID
    access_token: str | None = field(repr=False, default=None)
    refresh_token: str | None = field(repr=False, default=None)
    full_name: str | None = None
    first_name: str | None = None
    last_name: str | None = None
    email: str | None = None
    access_token_expires_at: datetime | None = None
    is_admin: bool = False

    @property
    def is_authenticated(self) -> bool:
        """Indicates whether the user has successfully logged in."""
        return self.id is not None

    @property
    def is_anonymous(self) -> bool:
        """Indicates whether the user is anonymous."""
        return isinstance(self, AnonymousAPIUser)

    def get_full_name(self) -> str | None:
        """Generate the closest thing to a full name if the full name field is not set."""
        full_name = self.full_name or " ".join(filter(None, (self.first_name, self.last_name)))
        if len(full_name) == 0:
            return None
        return full_name


@dataclass(kw_only=True, frozen=True)
class AuthenticatedAPIUser(APIUser):
    """The model for a an authenticated user of the API."""

    id: str
    email: str
    access_token: str = field(repr=False)
    refresh_token: str | None = field(default=None, repr=False)
    full_name: str | None = None
    first_name: str | None = None
    last_name: str | None = None


@dataclass(kw_only=True, frozen=True)
class AnonymousAPIUser(APIUser):
    """The model for an anonymous user of the API."""

    id: str
    is_admin: bool = field(init=False, default=False)

    @property
    def is_authenticated(self) -> bool:
        """We cannot authenticate anonymous users, so this is by definition False."""
        return False


class ServiceAdminId(StrEnum):
    """Types of internal service admins."""

    migrations = "migrations"
    secrets_rotation = "secrets_rotation"


@dataclass(kw_only=True, frozen=True)
class InternalServiceAdmin(APIUser):
    """Used to gain complete admin access by internal code components when performing tasks not started by users."""

    id: ServiceAdminId = ServiceAdminId.migrations
    access_token: str = field(repr=False, default="internal-service-admin", init=False)
    full_name: str | None = field(default=None, init=False)
    first_name: str | None = field(default=None, init=False)
    last_name: str | None = field(default=None, init=False)
    email: str | None = field(default=None, init=False)
    is_admin: bool = field(init=False, default=True)

    @property
    def is_authenticated(self) -> bool:
        """Internal admin users are always authenticated."""
        return True


class GitlabAccessLevel(Enum):
    """Gitlab access level for filtering projects."""

    PUBLIC = 1
    """User isn't a member but project is public"""
    MEMBER = 2
    """User is a member of the project"""
    ADMIN = 3
    """A user with at least DEVELOPER priviledges in gitlab is considered an Admin"""


class GitlabAPIProtocol(Protocol):
    """The interface for interacting with the Gitlab API."""

    async def filter_projects_by_access_level(
        self, user: APIUser, project_ids: list[str], min_access_level: GitlabAccessLevel
    ) -> list[str]:
        """Get a list of projects of which the user is a member with a specific access level."""
        ...

    async def get_project_url_from_path(self, user: APIUser, project_path: str) -> str | None:
        """Get the project ID from the path i.e. from /group1/subgroup2/project3."""
        ...


class UserStore(Protocol):
    """The interface through which Keycloak or a similar application can be accessed."""

    async def get_user_by_id(self, id: str, access_token: str) -> Optional[User]:
        """Get a user by their unique Keycloak user ID."""
        ...


@dataclass(frozen=True, eq=True, kw_only=True)
class User:
    """User model."""

    keycloak_id: str
    id: Optional[int] = None
    no_default_access: bool = False

    @classmethod
    def from_dict(cls, data: dict) -> User:
        """Create the model from a plain dictionary."""
        return cls(**data)


@dataclass(frozen=True, eq=True)
class Slug:
    """Slug used for namespaces, groups, projects, etc."""

    value: str
    # Slug regex rules
    # from https://docs.gitlab.com/ee/user/reserved_names.html#limitations-on-usernames-project-and-group-names
    # - cannot end in .git
    # - cannot end in .atom
    # - cannot contain any combination of two or more consecutive -._
    # - has to start with letter or number
    _regex: ClassVar[str] = r"^(?!.*\.git$|.*\.atom$|.*[\-._][\-._].*)[a-zA-Z0-9][a-zA-Z0-9\-_.]*$"

    def __init__(self, value: str) -> None:
        if not re.match(self._regex, value):
            raise errors.ValidationError(message=f"The slug {value} does not match the regex {self._regex}")
        object.__setattr__(self, "value", value)

    @classmethod
    def from_name(cls, name: str) -> Self:
        """Takes a name with any amount of invalid characters and transforms it in a valid slug."""
        lower_case = name.lower()
        no_space = re.sub(r"\s+", "-", lower_case)
        normalized = unicodedata.normalize("NFKD", no_space).encode("ascii", "ignore").decode("utf-8")
        valid_chars_pattern = [r"\w", ".", "_", "-"]
        no_invalid_characters = re.sub(f"[^{''.join(valid_chars_pattern)}]", "-", normalized)
        no_duplicates = re.sub(r"([._-])[._-]+", r"\1", no_invalid_characters)
        valid_start = re.sub(r"^[._-]", "", no_duplicates)
        valid_end = re.sub(r"[._-]$", "", valid_start)
        no_dot_git_or_dot_atom_at_end = re.sub(r"(\.atom|\.git)+$", "", valid_end)
        if len(no_dot_git_or_dot_atom_at_end) == 0:
            raise errors.ValidationError(
                message="The name for the project contains too many invalid characters so a slug could not be generated"
            )
        return cls(no_dot_git_or_dot_atom_at_end)

    @classmethod
    def from_user(cls, email: str | None, first_name: str | None, last_name: str | None, keycloak_id: str) -> Self:
        """Create a slug from a user."""
        if email:
            slug = email.split("@")[0]
        elif first_name and last_name:
            slug = first_name + "-" + last_name
        elif last_name:
            slug = last_name
        elif first_name:
            slug = first_name
        else:
            slug = "user_" + keycloak_id
        # The length limit is 99 but leave some space for modifications that may be added down the line
        # to filter out invalid characters or to generate a unique name
        slug = slug[:80]
        return cls.from_name(slug)

    def __str__(self) -> str:
        return self.value

    def __repr__(self) -> str:
        return self.value


class NamespaceSlug(Slug):
    """The slug for a group or user namespace."""


class ProjectSlug(Slug):
    """The slug for a project."""


class DataConnectorSlug(Slug):
    """The slug for a data connector."""


class __NamespaceCommonMixin:
    def __repr__(self) -> str:
        return "/".join([i.value for i in self.to_list()])

    def __getitem__(self, ind: int) -> Slug:
        return self.to_list()[ind]

    def __len__(self) -> int:
        return len(self.to_list())

    def to_list(self) -> list[Slug]:
        raise NotImplementedError

    def serialize(self) -> str:
        return "/".join([i.value for i in self.to_list()])


@dataclass(frozen=True, eq=True, repr=False)
class NamespacePath(__NamespaceCommonMixin):
    """The slug that makes up the path to a user or group namespace in Renku."""

    __match_args__ = ("first",)
    first: NamespaceSlug

    @overload
    def __truediv__(self, other: ProjectSlug) -> ProjectPath: ...
    @overload
    def __truediv__(self, other: DataConnectorSlug) -> DataConnectorPath: ...

    def __truediv__(self, other: ProjectSlug | DataConnectorSlug) -> ProjectPath | DataConnectorPath:
        """Create new entity path with an extra slug."""
        if isinstance(other, ProjectSlug):
            return ProjectPath(self.first, other)
        elif isinstance(other, DataConnectorSlug):
            return DataConnectorPath(self.first, other)
        else:
            raise errors.ProgrammingError(message=f"A path for a namespace cannot be further joined with {other}")

    def to_list(self) -> list[Slug]:
        """Convert to list of slugs."""
        return [self.first]

    def parent(self) -> Never:
        """The parent path."""
        raise errors.ProgrammingError(message="A namespace path has no parent")

    def last(self) -> NamespaceSlug:
        """Return the last slug in the path."""
        return self.first

    @classmethod
    def from_strings(cls, *slugs: str) -> Self:
        """Convert a string to a namespace path."""
        if len(slugs) != 1:
            raise errors.ValidationError(message=f"One slug string is needed to create a namespace path, got {slugs}.")
        return cls(NamespaceSlug(slugs[0]))


@dataclass(frozen=True, eq=True, repr=False)
class ProjectPath(__NamespaceCommonMixin):
    """The collection of slugs that makes up the path to a project in Renku."""

    __match_args__ = ("first", "second")
    first: NamespaceSlug
    second: ProjectSlug

    def __truediv__(self, other: DataConnectorSlug) -> DataConnectorInProjectPath:
        """Create new entity path with an extra slug."""
        if not isinstance(other, DataConnectorSlug):
            raise errors.ValidationError(
                message=f"A project path can only be joined with a data connector slug, but got {other}"
            )
        return DataConnectorInProjectPath(self.first, self.second, other)

    def to_list(self) -> list[Slug]:
        """Convert to list of slugs."""
        return [self.first, self.second]

    def parent(self) -> NamespacePath:
        """The parent path."""
        return NamespacePath(self.first)

    def last(self) -> ProjectSlug:
        """Return the last slug in the path."""
        return self.second

    @classmethod
    def from_strings(cls, *slugs: str) -> Self:
        """Convert strings to a project path."""
        if len(slugs) != 2:
            raise errors.ValidationError(message=f"Two slug strings are needed to create a project path, got {slugs}.")
        return cls(NamespaceSlug(slugs[0]), ProjectSlug(slugs[1]))


@dataclass(frozen=True, eq=True, repr=False)
class DataConnectorPath(__NamespaceCommonMixin):
    """The collection of slugs that makes up the path to a data connector in a user or group in Renku."""

    __match_args__ = ("first", "second")
    first: NamespaceSlug
    second: DataConnectorSlug

    def __truediv__(self, other: Never) -> Never:
        """Create new entity path with an extra slug."""
        raise errors.ProgrammingError(
            message="A path for a data connector in a user or group cannot be further joined with more slugs"
        )

    def to_list(self) -> list[Slug]:
        """Convert to list of slugs."""
        return [self.first, self.second]

    def parent(self) -> NamespacePath:
        """The parent path."""
        return NamespacePath(self.first)

    def last(self) -> DataConnectorSlug:
        """Return the last slug in the path."""
        return self.second

    @classmethod
    def from_strings(cls, *slugs: str) -> Self:
        """Convert strings to a data connector path."""
        if len(slugs) != 2:
            raise errors.ValidationError(
                message=f"Two slug strings are needed to create a data connector path, got {slugs}."
            )
        return cls(NamespaceSlug(slugs[0]), DataConnectorSlug(slugs[1]))


@dataclass(frozen=True, eq=True, repr=False)
class DataConnectorInProjectPath(__NamespaceCommonMixin):
    """The collection of slugs that makes up the path to a data connector in a projectj in Renku."""

    __match_args__ = ("first", "second", "third")
    first: NamespaceSlug
    second: ProjectSlug
    third: DataConnectorSlug

    def __truediv__(self, other: Never) -> Never:
        """Create new entity path with an extra slug."""
        raise errors.ProgrammingError(
            message="A path for a data connector in a project cannot be further joined with more slugs"
        )

    def to_list(self) -> list[Slug]:
        """Convert to list of slugs."""
        return [self.first, self.second, self.third]

    def parent(self) -> ProjectPath:
        """The parent path."""
        return ProjectPath(self.first, self.second)

    def last(self) -> DataConnectorSlug:
        """Return the last slug in the path."""
        return self.third

    @classmethod
    def from_strings(cls, *slugs: str) -> Self:
        """Convert strings to a data connector path."""
        if len(slugs) != 3:
            raise errors.ValidationError(
                message=f"Three slug strings are needed to create a data connector in project path, got {slugs}."
            )
        return cls(NamespaceSlug(slugs[0]), ProjectSlug(slugs[1]), DataConnectorSlug(slugs[2]))


AnyAPIUser = TypeVar("AnyAPIUser", bound=APIUser, covariant=True)


class Authenticator(Protocol[AnyAPIUser]):
    """Interface for authenticating users."""

    token_field: str

    async def authenticate(self, access_token: str, request: Request) -> AnyAPIUser:
        """Validates the user credentials (i.e. we can say that the user is a valid Renku user)."""
        ...


ResetType = NewType("ResetType", object)
"""This type represents that a value that may be None should be reset back to None or null.
This type should have only one instance, defined in the same file as this type.
"""

RESET: ResetType = ResetType(object())
"""The single instance of the ResetType, can be compared to similar to None, i.e. `if value is RESET`"""


class ResourceType(StrEnum):
    """All possible resources stored in Authzed."""

    project = "project"
    user = "user"
    anonymous_user = "anonymous_user"
    platform = "platform"
    group = "group"
    user_namespace = "user_namespace"
    data_connector = "data_connector"

1	"""Base models shared by services."""
2
3	from __future__ import annotations	2✔
4
5	import re	2✔
6	import unicodedata	2✔
7	from dataclasses import dataclass, field	2✔
8	from datetime import datetime	2✔
9	from enum import Enum, StrEnum	2✔
10	from typing import ClassVar, Never, NewType, Optional, Protocol, Self, TypeVar, overload	2✔
11
12	from sanic import Request	2✔
13
14	from renku_data_services.errors import errors	2✔
15
16
17	@dataclass(kw_only=True, frozen=True)	2✔
18	class APIUser:	2✔
19	"""The model for a user of the API, used for authentication."""
20
21	id: str \| None = None # the sub claim in the access token - i.e. the Keycloak user ID	2✔
22	access_token: str \| None = field(repr=False, default=None)	2✔
23	refresh_token: str \| None = field(repr=False, default=None)	2✔
24	full_name: str \| None = None	2✔
25	first_name: str \| None = None	2✔
26	last_name: str \| None = None	2✔
27	email: str \| None = None	2✔
28	access_token_expires_at: datetime \| None = None	2✔
29	is_admin: bool = False	2✔
30
31	@property	2✔
32	def is_authenticated(self) -> bool:	2✔
33	"""Indicates whether the user has successfully logged in."""
34	return self.id is not None	2✔
35
36	@property	2✔
37	def is_anonymous(self) -> bool:	2✔
38	"""Indicates whether the user is anonymous."""
39	return isinstance(self, AnonymousAPIUser)	1✔
40
41	def get_full_name(self) -> str \| None:	2✔
42	"""Generate the closest thing to a full name if the full name field is not set."""
43	full_name = self.full_name or " ".join(filter(None, (self.first_name, self.last_name)))	1✔
44	if len(full_name) == 0:	1✔
45	return None	×
46	return full_name	1✔
47
48
49	@dataclass(kw_only=True, frozen=True)	2✔
50	class AuthenticatedAPIUser(APIUser):	2✔
51	"""The model for a an authenticated user of the API."""
52
53	id: str	2✔
54	email: str	2✔
55	access_token: str = field(repr=False)	2✔
56	refresh_token: str \| None = field(default=None, repr=False)	2✔
57	full_name: str \| None = None	2✔
58	first_name: str \| None = None	2✔
59	last_name: str \| None = None	2✔
60
61
62	@dataclass(kw_only=True, frozen=True)	2✔
63	class AnonymousAPIUser(APIUser):	2✔
64	"""The model for an anonymous user of the API."""
65
66	id: str	2✔
67	is_admin: bool = field(init=False, default=False)	2✔
68
69	@property	2✔
70	def is_authenticated(self) -> bool:	2✔
71	"""We cannot authenticate anonymous users, so this is by definition False."""
72	return False	1✔
73
74
75	class ServiceAdminId(StrEnum):	2✔
76	"""Types of internal service admins."""
77
78	migrations = "migrations"	2✔
79	secrets_rotation = "secrets_rotation"	2✔
80
81
82	@dataclass(kw_only=True, frozen=True)	2✔
83	class InternalServiceAdmin(APIUser):	2✔
84	"""Used to gain complete admin access by internal code components when performing tasks not started by users."""
85
86	id: ServiceAdminId = ServiceAdminId.migrations	2✔
87	access_token: str = field(repr=False, default="internal-service-admin", init=False)	2✔
88	full_name: str \| None = field(default=None, init=False)	2✔
89	first_name: str \| None = field(default=None, init=False)	2✔
90	last_name: str \| None = field(default=None, init=False)	2✔
91	email: str \| None = field(default=None, init=False)	2✔
92	is_admin: bool = field(init=False, default=True)	2✔
93
94	@property	2✔
95	def is_authenticated(self) -> bool:	2✔
96	"""Internal admin users are always authenticated."""
97	return True	×
98
99
100	class GitlabAccessLevel(Enum):	2✔
101	"""Gitlab access level for filtering projects."""
102
103	PUBLIC = 1	2✔
104	"""User isn't a member but project is public"""	2✔
105	MEMBER = 2	2✔
106	"""User is a member of the project"""	2✔
107	ADMIN = 3	2✔
108	"""A user with at least DEVELOPER priviledges in gitlab is considered an Admin"""	2✔
109
110
111	class GitlabAPIProtocol(Protocol):	2✔
112	"""The interface for interacting with the Gitlab API."""
113
114	async def filter_projects_by_access_level(	2✔
115	self, user: APIUser, project_ids: list[str], min_access_level: GitlabAccessLevel
116	) -> list[str]:
117	"""Get a list of projects of which the user is a member with a specific access level."""
118	...	×
119
120	async def get_project_url_from_path(self, user: APIUser, project_path: str) -> str \| None:	2✔
121	"""Get the project ID from the path i.e. from /group1/subgroup2/project3."""
NEW 122	...	×
123
124
125	class UserStore(Protocol):	2✔
126	"""The interface through which Keycloak or a similar application can be accessed."""
127
128	async def get_user_by_id(self, id: str, access_token: str) -> Optional[User]:	2✔
129	"""Get a user by their unique Keycloak user ID."""
130	...	×
131
132
133	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
134	class User:	2✔
135	"""User model."""
136
137	keycloak_id: str	2✔
138	id: Optional[int] = None	2✔
139	no_default_access: bool = False	2✔
140
141	@classmethod	2✔
142	def from_dict(cls, data: dict) -> User:	2✔
143	"""Create the model from a plain dictionary."""
144	return cls(**data)	×
145
146
147	@dataclass(frozen=True, eq=True)	2✔
148	class Slug:	2✔
149	"""Slug used for namespaces, groups, projects, etc."""
150
151	value: str	2✔
152	# Slug regex rules
153	# from https://docs.gitlab.com/ee/user/reserved_names.html#limitations-on-usernames-project-and-group-names
154	# - cannot end in .git
155	# - cannot end in .atom
156	# - cannot contain any combination of two or more consecutive -._
157	# - has to start with letter or number
158	_regex: ClassVar[str] = r"^(?!.\.git$\|.\.atom$\|.[\-._][\-._].)[a-zA-Z0-9][a-zA-Z0-9\-_.]*$"	2✔
159
160	def __init__(self, value: str) -> None:	2✔
161	if not re.match(self._regex, value):	2✔
162	raise errors.ValidationError(message=f"The slug {value} does not match the regex {self._regex}")	2✔
163	object.__setattr__(self, "value", value)	2✔
164
165	@classmethod	2✔
166	def from_name(cls, name: str) -> Self:	2✔
167	"""Takes a name with any amount of invalid characters and transforms it in a valid slug."""
168	lower_case = name.lower()	2✔
169	no_space = re.sub(r"\s+", "-", lower_case)	2✔
170	normalized = unicodedata.normalize("NFKD", no_space).encode("ascii", "ignore").decode("utf-8")	2✔
171	valid_chars_pattern = [r"\w", ".", "_", "-"]	2✔
172	no_invalid_characters = re.sub(f"[^{''.join(valid_chars_pattern)}]", "-", normalized)	2✔
173	no_duplicates = re.sub(r"([._-])[._-]+", r"\1", no_invalid_characters)	2✔
174	valid_start = re.sub(r"^[._-]", "", no_duplicates)	2✔
175	valid_end = re.sub(r"[._-]$", "", valid_start)	2✔
176	no_dot_git_or_dot_atom_at_end = re.sub(r"(\.atom\|\.git)+$", "", valid_end)	2✔
177	if len(no_dot_git_or_dot_atom_at_end) == 0:	2✔
178	raise errors.ValidationError(	2✔
179	message="The name for the project contains too many invalid characters so a slug could not be generated"
180	)
181	return cls(no_dot_git_or_dot_atom_at_end)	2✔
182
183	@classmethod	2✔
184	def from_user(cls, email: str \| None, first_name: str \| None, last_name: str \| None, keycloak_id: str) -> Self:	2✔
185	"""Create a slug from a user."""
186	if email:	2✔
187	slug = email.split("@")[0]	2✔
188	elif first_name and last_name:	×
189	slug = first_name + "-" + last_name	×
190	elif last_name:	×
191	slug = last_name	×
192	elif first_name:	×
193	slug = first_name	×
194	else:
195	slug = "user_" + keycloak_id	×
196	# The length limit is 99 but leave some space for modifications that may be added down the line
197	# to filter out invalid characters or to generate a unique name
198	slug = slug[:80]	2✔
199	return cls.from_name(slug)	2✔
200
201	def __str__(self) -> str:	2✔
202	return self.value	2✔
203
204	def __repr__(self) -> str:	2✔
205	return self.value	×
206
207
208	class NamespaceSlug(Slug):	2✔
209	"""The slug for a group or user namespace."""
210
211
212	class ProjectSlug(Slug):	2✔
213	"""The slug for a project."""
214
215
216	class DataConnectorSlug(Slug):	2✔
217	"""The slug for a data connector."""
218
219
220	class __NamespaceCommonMixin:	2✔
221	def __repr__(self) -> str:	2✔
222	return "/".join([i.value for i in self.to_list()])	2✔
223
224	def __getitem__(self, ind: int) -> Slug:	2✔
225	return self.to_list()[ind]	×
226
227	def __len__(self) -> int:	2✔
228	return len(self.to_list())	×
229
230	def to_list(self) -> list[Slug]:	2✔
231	raise NotImplementedError	×
232
233	def serialize(self) -> str:	2✔
234	return "/".join([i.value for i in self.to_list()])	2✔
235
236
237	@dataclass(frozen=True, eq=True, repr=False)	2✔
238	class NamespacePath(__NamespaceCommonMixin):	2✔
239	"""The slug that makes up the path to a user or group namespace in Renku."""
240
241	__match_args__ = ("first",)	2✔
242	first: NamespaceSlug	2✔
243
244	@overload	2✔
245	def __truediv__(self, other: ProjectSlug) -> ProjectPath: ...	2✔
246	@overload	2✔
247	def __truediv__(self, other: DataConnectorSlug) -> DataConnectorPath: ...	2✔
248
249	def __truediv__(self, other: ProjectSlug \| DataConnectorSlug) -> ProjectPath \| DataConnectorPath:	2✔
250	"""Create new entity path with an extra slug."""
251	if isinstance(other, ProjectSlug):	1✔
252	return ProjectPath(self.first, other)	1✔
253	elif isinstance(other, DataConnectorSlug):	1✔
254	return DataConnectorPath(self.first, other)	1✔
255	else:
256	raise errors.ProgrammingError(message=f"A path for a namespace cannot be further joined with {other}")	×
257
258	def to_list(self) -> list[Slug]:	2✔
259	"""Convert to list of slugs."""
260	return [self.first]	2✔
261
262	def parent(self) -> Never:	2✔
263	"""The parent path."""
264	raise errors.ProgrammingError(message="A namespace path has no parent")	×
265
266	def last(self) -> NamespaceSlug:	2✔
267	"""Return the last slug in the path."""
268	return self.first	×
269
270	@classmethod	2✔
271	def from_strings(cls, *slugs: str) -> Self:	2✔
272	"""Convert a string to a namespace path."""
273	if len(slugs) != 1:	2✔
274	raise errors.ValidationError(message=f"One slug string is needed to create a namespace path, got {slugs}.")	×
275	return cls(NamespaceSlug(slugs[0]))	2✔
276
277
278	@dataclass(frozen=True, eq=True, repr=False)	2✔
279	class ProjectPath(__NamespaceCommonMixin):	2✔
280	"""The collection of slugs that makes up the path to a project in Renku."""
281
282	__match_args__ = ("first", "second")	2✔
283	first: NamespaceSlug	2✔
284	second: ProjectSlug	2✔
285
286	def __truediv__(self, other: DataConnectorSlug) -> DataConnectorInProjectPath:	2✔
287	"""Create new entity path with an extra slug."""
288	if not isinstance(other, DataConnectorSlug):	1✔
289	raise errors.ValidationError(	×
290	message=f"A project path can only be joined with a data connector slug, but got {other}"
291	)
292	return DataConnectorInProjectPath(self.first, self.second, other)	1✔
293
294	def to_list(self) -> list[Slug]:	2✔
295	"""Convert to list of slugs."""
296	return [self.first, self.second]	2✔
297
298	def parent(self) -> NamespacePath:	2✔
299	"""The parent path."""
300	return NamespacePath(self.first)	×
301
302	def last(self) -> ProjectSlug:	2✔
303	"""Return the last slug in the path."""
304	return self.second	×
305
306	@classmethod	2✔
307	def from_strings(cls, *slugs: str) -> Self:	2✔
308	"""Convert strings to a project path."""
309	if len(slugs) != 2:	2✔
310	raise errors.ValidationError(message=f"Two slug strings are needed to create a project path, got {slugs}.")	×
311	return cls(NamespaceSlug(slugs[0]), ProjectSlug(slugs[1]))	2✔
312
313
314	@dataclass(frozen=True, eq=True, repr=False)	2✔
315	class DataConnectorPath(__NamespaceCommonMixin):	2✔
316	"""The collection of slugs that makes up the path to a data connector in a user or group in Renku."""
317
318	__match_args__ = ("first", "second")	2✔
319	first: NamespaceSlug	2✔
320	second: DataConnectorSlug	2✔
321
322	def __truediv__(self, other: Never) -> Never:	2✔
323	"""Create new entity path with an extra slug."""
324	raise errors.ProgrammingError(	×
325	message="A path for a data connector in a user or group cannot be further joined with more slugs"
326	)
327
328	def to_list(self) -> list[Slug]:	2✔
329	"""Convert to list of slugs."""
330	return [self.first, self.second]	2✔
331
332	def parent(self) -> NamespacePath:	2✔
333	"""The parent path."""
334	return NamespacePath(self.first)	2✔
335
336	def last(self) -> DataConnectorSlug:	2✔
337	"""Return the last slug in the path."""
338	return self.second	2✔
339
340	@classmethod	2✔
341	def from_strings(cls, *slugs: str) -> Self:	2✔
342	"""Convert strings to a data connector path."""
343	if len(slugs) != 2:	2✔
344	raise errors.ValidationError(	×
345	message=f"Two slug strings are needed to create a data connector path, got {slugs}."
346	)
347	return cls(NamespaceSlug(slugs[0]), DataConnectorSlug(slugs[1]))	2✔
348
349
350	@dataclass(frozen=True, eq=True, repr=False)	2✔
351	class DataConnectorInProjectPath(__NamespaceCommonMixin):	2✔
352	"""The collection of slugs that makes up the path to a data connector in a projectj in Renku."""
353
354	__match_args__ = ("first", "second", "third")	2✔
355	first: NamespaceSlug	2✔
356	second: ProjectSlug	2✔
357	third: DataConnectorSlug	2✔
358
359	def __truediv__(self, other: Never) -> Never:	2✔
360	"""Create new entity path with an extra slug."""
361	raise errors.ProgrammingError(	×
362	message="A path for a data connector in a project cannot be further joined with more slugs"
363	)
364
365	def to_list(self) -> list[Slug]:	2✔
366	"""Convert to list of slugs."""
367	return [self.first, self.second, self.third]	2✔
368
369	def parent(self) -> ProjectPath:	2✔
370	"""The parent path."""
371	return ProjectPath(self.first, self.second)	2✔
372
373	def last(self) -> DataConnectorSlug:	2✔
374	"""Return the last slug in the path."""
375	return self.third	2✔
376
377	@classmethod	2✔
378	def from_strings(cls, *slugs: str) -> Self:	2✔
379	"""Convert strings to a data connector path."""
380	if len(slugs) != 3:	1✔
381	raise errors.ValidationError(	×
382	message=f"Three slug strings are needed to create a data connector in project path, got {slugs}."
383	)
384	return cls(NamespaceSlug(slugs[0]), ProjectSlug(slugs[1]), DataConnectorSlug(slugs[2]))	1✔
385
386
387	AnyAPIUser = TypeVar("AnyAPIUser", bound=APIUser, covariant=True)	2✔
388
389
390	class Authenticator(Protocol[AnyAPIUser]):	2✔
391	"""Interface for authenticating users."""
392
393	token_field: str	2✔
394
395	async def authenticate(self, access_token: str, request: Request) -> AnyAPIUser:	2✔
396	"""Validates the user credentials (i.e. we can say that the user is a valid Renku user)."""
397	...	×
398
399
400	ResetType = NewType("ResetType", object)	2✔
401	"""This type represents that a value that may be None should be reset back to None or null.	2✔
402	This type should have only one instance, defined in the same file as this type.
403	"""
404
405	RESET: ResetType = ResetType(object())	2✔
406	"""The single instance of the ResetType, can be compared to similar to None, i.e. `if value is RESET`"""	2✔
407
408
409	class ResourceType(StrEnum):	2✔
410	"""All possible resources stored in Authzed."""
411
412	project = "project"	2✔
413	user = "user"	2✔
414	anonymous_user = "anonymous_user"	2✔
415	platform = "platform"	2✔
416	group = "group"	2✔
417	user_namespace = "user_namespace"	2✔
418	data_connector = "data_connector"	2✔

SwissDataScienceCenter / renku-data-services / 14377770991

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