19267889034

Committed 11 Nov 2025 01:56PM UTC coverage: 86.481% (+0.1%) from 86.352%

Build # 19267889034

Build Type

Pull #1095

github

Committed by

web-flow

Commit Message

Merge 1da835101 into ffc3f8457

Pull Request Pull Request #1095: Feat: Improve repository response

Run Details

265 of 298 new or added lines in 7 files covered. (88.93%)

4 existing lines in 3 files now uncovered.

23022 of 26621 relevant lines covered (86.48%)

1.52 hits per line

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

96.88

/components/renku_data_services/repositories/models.py

"""Models for connected services."""

from __future__ import annotations

import dataclasses
from dataclasses import dataclass
from enum import StrEnum
from typing import Literal

from ulid import ULID

from renku_data_services.connected_services.orm import OAuth2ClientORM, OAuth2ConnectionORM
from renku_data_services.repositories.git_url import GitUrlError


@dataclass(frozen=True, eq=True, kw_only=True)
class RepositoryPermissions:
    """Repository permissions for git operations."""

    pull: bool
    push: bool

    @classmethod
    def default(cls) -> RepositoryPermissions:
        """Default permissions."""
        return cls(pull=False, push=False)


@dataclass(frozen=True, eq=True, kw_only=True)
class RepositoryMetadata:
    """Repository metadata."""

    etag: str | None
    git_http_url: str
    web_url: str
    permissions: RepositoryPermissions


@dataclass(frozen=True, eq=True, kw_only=True)
class RepositoryProviderMatch:
    """Repository provider match data."""

    provider_id: str
    connection_id: ULID | None
    repository_metadata: RepositoryMetadata | None


@dataclass(frozen=True, eq=True, kw_only=True)
class ProviderConnection:
    """Repository connection data."""

    id: ULID
    provider_id: str
    status: str

    @classmethod
    def fromORM(cls, e: OAuth2ConnectionORM) -> ProviderConnection:
        """Create a ProviderConnection from an ORM class."""
        return ProviderConnection(id=e.id, provider_id=e.client_id, status=e.status)


@dataclass(frozen=True, eq=True, kw_only=True)
class ProviderData:
    """Repository provider data."""

    id: str
    name: str
    url: str

    @classmethod
    def fromORM(cls, e: OAuth2ClientORM) -> ProviderData:
        """Create ProviderData from an ORM class."""
        return ProviderData(id=e.id, name=e.display_name, url=e.url)


@dataclass(frozen=True, eq=True, kw_only=True)
class RepositoryProviderData:
    """Repository provider match data."""

    provider: ProviderData
    connection: ProviderConnection | None
    repository_metadata: RepositoryMetadata | None


@dataclass(frozen=True, eq=True, kw_only=True)
class Metadata:
    """Metadata about a repository."""

    etag: str | None = None
    git_url: str
    web_url: str | None = None
    pull_permission: bool
    push_permission: bool | None = None

    @classmethod
    def fromRepoMeta(cls, rm: RepositoryMetadata) -> Metadata:
        """Create Metadata from RepositoryMetadata."""
        return Metadata(
            etag=rm.etag,
            git_url=rm.git_http_url,
            web_url=rm.web_url,
            pull_permission=rm.permissions.pull,
            push_permission=rm.permissions.push,
        )


class RepositoryMetadataError(StrEnum):
    """Possible errors when retrieving repository metadata."""

    metadata_unauthorized = "metadata_unauthorized"
    metadata_validation = "metadata_validation"
    metadata_unknown = "metadata_unknown"


type RepositoryError = GitUrlError | RepositoryMetadataError


@dataclass(frozen=True, eq=True, kw_only=True)
class RepositoryDataResult:
    """Information when retrieving a repository."""

    provider: ProviderData | None = None
    connection: ProviderConnection | None = None
    error: RepositoryError | None = None
    metadata: Metadata | Literal["Unmodified"] | None = None

    def with_metadata(
        self, rm: Metadata | RepositoryMetadata | Literal["304"] | RepositoryError
    ) -> RepositoryDataResult:
        """Return a new result with metadatat set."""
        match rm:
            case "304":
                return dataclasses.replace(self, metadata="Unmodified")
            case GitUrlError() as err:
                return self.with_error(err)
            case RepositoryMetadataError() as err:
                return self.with_error(err)
            case Metadata() as md:
                return dataclasses.replace(self, metadata=md)
            case RepositoryMetadata() as md:
                return dataclasses.replace(self, metadata=Metadata.fromRepoMeta(md))

    def with_provider(self, p: ProviderData | None) -> RepositoryDataResult:
        """Return a new result with the provider set."""
        return dataclasses.replace(self, provider=p)

    def with_provider_orm(self, p: OAuth2ClientORM | None) -> RepositoryDataResult:
        """Return a new result with the provider set."""
        return self.with_provider(ProviderData.fromORM(p) if p else None)

    def with_connection(self, c: ProviderConnection | None) -> RepositoryDataResult:
        """Return a new result with the connection set."""
        return dataclasses.replace(self, connection=c)

    def with_connection_orm(self, c: OAuth2ConnectionORM | None) -> RepositoryDataResult:
        """Return a new result with the connection set."""
        return self.with_connection(ProviderConnection.fromORM(c) if c else None)

    def with_error(self, err: RepositoryError | None) -> RepositoryDataResult:
        """Return a new result with the error set."""
        return dataclasses.replace(self, error=err)

    @property
    def is_error(self) -> bool:
        """Return whether this is an error result."""
        return self.error is not None

    @property
    def is_success(self) -> bool:
        """Return whether this is a success result."""
        return not self.error and self.metadata is not None

1	"""Models for connected services."""
2
3	from __future__ import annotations	2✔
4
5	import dataclasses	2✔
6	from dataclasses import dataclass	2✔
7	from enum import StrEnum	2✔
8	from typing import Literal	2✔
9
10	from ulid import ULID	2✔
11
12	from renku_data_services.connected_services.orm import OAuth2ClientORM, OAuth2ConnectionORM	2✔
13	from renku_data_services.repositories.git_url import GitUrlError	2✔
14
15
16	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
17	class RepositoryPermissions:	2✔
18	"""Repository permissions for git operations."""
19
20	pull: bool	2✔
21	push: bool	2✔
22
23	@classmethod	2✔
24	def default(cls) -> RepositoryPermissions:	2✔
25	"""Default permissions."""
26	return cls(pull=False, push=False)	×
27
28
29	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
30	class RepositoryMetadata:	2✔
31	"""Repository metadata."""
32
33	etag: str \| None	2✔
34	git_http_url: str	2✔
35	web_url: str	2✔
36	permissions: RepositoryPermissions	2✔
37
38
39	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
40	class RepositoryProviderMatch:	2✔
41	"""Repository provider match data."""
42
43	provider_id: str	2✔
44	connection_id: ULID \| None	2✔
45	repository_metadata: RepositoryMetadata \| None	2✔
46
47
48	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
49	class ProviderConnection:	2✔
50	"""Repository connection data."""
51
52	id: ULID	2✔
53	provider_id: str	2✔
54	status: str	2✔
55
56	@classmethod	2✔
57	def fromORM(cls, e: OAuth2ConnectionORM) -> ProviderConnection:	2✔
58	"""Create a ProviderConnection from an ORM class."""
59	return ProviderConnection(id=e.id, provider_id=e.client_id, status=e.status)	1✔
60
61
62	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
63	class ProviderData:	2✔
64	"""Repository provider data."""
65
66	id: str	2✔
67	name: str	2✔
68	url: str	2✔
69
70	@classmethod	2✔
71	def fromORM(cls, e: OAuth2ClientORM) -> ProviderData:	2✔
72	"""Create ProviderData from an ORM class."""
73	return ProviderData(id=e.id, name=e.display_name, url=e.url)	1✔
74
75
76	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
77	class RepositoryProviderData:	2✔
78	"""Repository provider match data."""
79
80	provider: ProviderData	2✔
81	connection: ProviderConnection \| None	2✔
82	repository_metadata: RepositoryMetadata \| None	2✔
83
84
85	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
86	class Metadata:	2✔
87	"""Metadata about a repository."""
88
89	etag: str \| None = None	2✔
90	git_url: str	2✔
91	web_url: str \| None = None	2✔
92	pull_permission: bool	2✔
93	push_permission: bool \| None = None	2✔
94
95	@classmethod	2✔
96	def fromRepoMeta(cls, rm: RepositoryMetadata) -> Metadata:	2✔
97	"""Create Metadata from RepositoryMetadata."""
98	return Metadata(	1✔
99	etag=rm.etag,
100	git_url=rm.git_http_url,
101	web_url=rm.web_url,
102	pull_permission=rm.permissions.pull,
103	push_permission=rm.permissions.push,
104	)
105
106
107	class RepositoryMetadataError(StrEnum):	2✔
108	"""Possible errors when retrieving repository metadata."""
109
110	metadata_unauthorized = "metadata_unauthorized"	2✔
111	metadata_validation = "metadata_validation"	2✔
112	metadata_unknown = "metadata_unknown"	2✔
113
114
115	type RepositoryError = GitUrlError \| RepositoryMetadataError	2✔
116
117
118	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
119	class RepositoryDataResult:	2✔
120	"""Information when retrieving a repository."""
121
122	provider: ProviderData \| None = None	2✔
123	connection: ProviderConnection \| None = None	2✔
124	error: RepositoryError \| None = None	2✔
125	metadata: Metadata \| Literal["Unmodified"] \| None = None	2✔
126
127	def with_metadata(	2✔
128	self, rm: Metadata \| RepositoryMetadata \| Literal["304"] \| RepositoryError
129	) -> RepositoryDataResult:
130	"""Return a new result with metadatat set."""
131	match rm:	1✔
132	case "304":	1✔
NEW 133	return dataclasses.replace(self, metadata="Unmodified")	×
134	case GitUrlError() as err:	1✔
NEW 135	return self.with_error(err)	×
136	case RepositoryMetadataError() as err:	1✔
137	return self.with_error(err)	1✔
138	case Metadata() as md:	1✔
139	return dataclasses.replace(self, metadata=md)	1✔
140	case RepositoryMetadata() as md:	1✔
141	return dataclasses.replace(self, metadata=Metadata.fromRepoMeta(md))	1✔
142
143	def with_provider(self, p: ProviderData \| None) -> RepositoryDataResult:	2✔
144	"""Return a new result with the provider set."""
145	return dataclasses.replace(self, provider=p)	1✔
146
147	def with_provider_orm(self, p: OAuth2ClientORM \| None) -> RepositoryDataResult:	2✔
148	"""Return a new result with the provider set."""
149	return self.with_provider(ProviderData.fromORM(p) if p else None)	1✔
150
151	def with_connection(self, c: ProviderConnection \| None) -> RepositoryDataResult:	2✔
152	"""Return a new result with the connection set."""
153	return dataclasses.replace(self, connection=c)	1✔
154
155	def with_connection_orm(self, c: OAuth2ConnectionORM \| None) -> RepositoryDataResult:	2✔
156	"""Return a new result with the connection set."""
157	return self.with_connection(ProviderConnection.fromORM(c) if c else None)	1✔
158
159	def with_error(self, err: RepositoryError \| None) -> RepositoryDataResult:	2✔
160	"""Return a new result with the error set."""
161	return dataclasses.replace(self, error=err)	1✔
162
163	@property	2✔
164	def is_error(self) -> bool:	2✔
165	"""Return whether this is an error result."""
166	return self.error is not None	2✔
167
168	@property	2✔
169	def is_success(self) -> bool:	2✔
170	"""Return whether this is a success result."""
171	return not self.error and self.metadata is not None	2✔

SwissDataScienceCenter / renku-data-services / 19267889034

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