17404558887

Committed 02 Sep 2025 01:10PM UTC coverage: 86.992% (+0.02%) from 86.968%

Build # 17404558887

Build Type

push

github

Committed by

web-flow

Commit Message

feat: decouple gitlab from data service (#923)

* refactor: delete unsed code, clean up internal gitlab usage

* feat: allow disabling internal gitlab repositories

Run Details

15 of 39 new or added lines in 8 files covered. (38.46%)

3 existing lines in 3 files now uncovered.

21768 of 25023 relevant lines covered (86.99%)

1.53 hits per line

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

83.58

/components/renku_data_services/git/gitlab.py

"""Gitlab API."""

from dataclasses import dataclass, field
from typing import Any, cast

import httpx

from renku_data_services.base_models import APIUser, GitlabAccessLevel
from renku_data_services.errors import errors
from renku_data_services.utils.core import get_ssl_context


@dataclass(kw_only=True)
class GitlabAPI:
    """Adapter for interacting with the gitlab API."""

    gitlab_url: str
    gitlab_graphql_url: str = field(init=False)

    def __post_init__(self) -> None:
        """Sets the graphql url for gitlab."""
        gitlab_url = self.gitlab_url

        if not gitlab_url.startswith("http") and "://" not in gitlab_url:
            raise errors.ConfigurationError(message=f"Gitlab URL should start with 'http(s)://', got: {gitlab_url}")

        gitlab_url = gitlab_url.rstrip("/")

        self.gitlab_graphql_url = f"{gitlab_url}/api/graphql"

    async def filter_projects_by_access_level(
        self, user: APIUser, project_ids: list[str], min_access_level: GitlabAccessLevel
    ) -> list[str]:
        """Filter projects this user can access in gitlab with at least access level."""

        if not user.access_token or not user.full_name:
            return []
        header = {"Authorization": f"Bearer {user.access_token}", "Content-Type": "application/json"}
        ids = ",".join(f'"gid://gitlab/Project/{id}"' for id in project_ids)
        query_body = f"""
                    pageInfo {{
                      hasNextPage
                    }}
                    nodes {{
                        id
                        projectMembers(search: "{user.full_name}") {{
                            nodes {{
                                user {{
                                    id
                                    name
                                }}
                                accessLevel {{
                                    integerValue
                                }}
                            }}
                        }}
                    }}
        """
        body = {
            "query": f"""{{
                projects(ids: [{ids}]) {{
                    {query_body}
                }}
            }}
            """
        }

        async def _query_gitlab_graphql(body: dict[str, Any], header: dict[str, Any]) -> dict[str, Any]:
            async with httpx.AsyncClient(verify=get_ssl_context(), timeout=5) as client:
                resp = await client.post(self.gitlab_graphql_url, json=body, headers=header, timeout=10)
            if resp.status_code != 200:
                raise errors.BaseError(message=f"Error querying Gitlab api {self.gitlab_graphql_url}: {resp.text}")
            result = cast(dict[str, Any], resp.json())

            if "data" not in result or "projects" not in result["data"]:
                raise errors.BaseError(message=f"Got unexpected response from Gitlab: {result}")
            return result

        resp_body = await _query_gitlab_graphql(body, header)
        result: list[str] = []

        def _process_projects(
            resp_body: dict[str, Any], min_access_level: GitlabAccessLevel, result: list[str]
        ) -> None:
            for project in resp_body["data"]["projects"]["nodes"]:
                if min_access_level != GitlabAccessLevel.PUBLIC:
                    if not project["projectMembers"]["nodes"]:
                        continue
                    if min_access_level == GitlabAccessLevel.ADMIN:
                        max_level = max(
                            n["accessLevel"]["integerValue"]
                            for n in project["projectMembers"]["nodes"]
                            if n["user"]["id"].rsplit("/", maxsplit=1)[-1] == user.id
                        )
                        if max_level < 30:
                            continue
                result.append(project["id"].rsplit("/", maxsplit=1)[-1])

        _process_projects(resp_body, min_access_level, result)
        page_info = resp_body["data"]["projects"]["pageInfo"]
        while page_info["hasNextPage"]:
            cursor = page_info["endCursor"]
            body = {
                "query": f"""{{
                    projects(ids: [{ids}], after: "{cursor}") {{
                        {query_body}
                    }}
                }}
                """
            }
            resp_body = await _query_gitlab_graphql(body, header)
            page_info = resp_body["data"]["projects"]["pageInfo"]
            _process_projects(resp_body, min_access_level, result)

        return result


@dataclass(kw_only=True)
class EmptyGitlabAPI:
    """An empty gitlab API used to decouple gitlab from Renku."""

    async def filter_projects_by_access_level(self, _: APIUser, __: list[str], ___: GitlabAccessLevel) -> list[str]:
        """Always return an empty list."""
        return []


@dataclass(kw_only=True)
class DummyGitlabAPI:
    """Dummy gitlab API.

    The user with name Admin Doe has admin access to project 123456 and member access to 999999.
    """

    _store = {
        "Admin Doe": {
            GitlabAccessLevel.MEMBER: ["999999", "123456"],
            GitlabAccessLevel.ADMIN: ["123456"],
        },
    }

    async def filter_projects_by_access_level(
        self, user: APIUser, project_ids: list[str], min_access_level: GitlabAccessLevel
    ) -> list[str]:
        """Filter projects this user can access in gitlab with at least access level."""
        if not user.access_token or not user.full_name:
            return []
        if min_access_level == GitlabAccessLevel.PUBLIC:
            return []
        user_projects = self._store.get(user.full_name, {}).get(min_access_level, [])
        return [p for p in project_ids if p in user_projects]

1	"""Gitlab API."""
2
3	from dataclasses import dataclass, field	2✔
4	from typing import Any, cast	2✔
5
6	import httpx	2✔
7
8	from renku_data_services.base_models import APIUser, GitlabAccessLevel	2✔
9	from renku_data_services.errors import errors	2✔
10	from renku_data_services.utils.core import get_ssl_context	2✔
11
12
13	@dataclass(kw_only=True)	2✔
14	class GitlabAPI:	2✔
15	"""Adapter for interacting with the gitlab API."""
16
17	gitlab_url: str	2✔
18	gitlab_graphql_url: str = field(init=False)	2✔
19
20	def __post_init__(self) -> None:	2✔
21	"""Sets the graphql url for gitlab."""
22	gitlab_url = self.gitlab_url	1✔
23
24	if not gitlab_url.startswith("http") and "://" not in gitlab_url:	1✔
25	raise errors.ConfigurationError(message=f"Gitlab URL should start with 'http(s)://', got: {gitlab_url}")	×
26
27	gitlab_url = gitlab_url.rstrip("/")	1✔
28
29	self.gitlab_graphql_url = f"{gitlab_url}/api/graphql"	1✔
30
31	async def filter_projects_by_access_level(	2✔
32	self, user: APIUser, project_ids: list[str], min_access_level: GitlabAccessLevel
33	) -> list[str]:
34	"""Filter projects this user can access in gitlab with at least access level."""
35
36	if not user.access_token or not user.full_name:	1✔
37	return []	×
38	header = {"Authorization": f"Bearer {user.access_token}", "Content-Type": "application/json"}	1✔
39	ids = ",".join(f'"gid://gitlab/Project/{id}"' for id in project_ids)	1✔
40	query_body = f"""	1✔
41	pageInfo {{
42	hasNextPage
43	}}
44	nodes {{
45	id
46	projectMembers(search: "{user.full_name}") {{
47	nodes {{
48	user {{
49	id
50	name
51	}}
52	accessLevel {{
53	integerValue
54	}}
55	}}
56	}}
57	}}
58	"""
59	body = {	1✔
60	"query": f"""{{
61	projects(ids: [{ids}]) {{
62	{query_body}
63	}}
64	}}
65	"""
66	}
67
68	async def _query_gitlab_graphql(body: dict[str, Any], header: dict[str, Any]) -> dict[str, Any]:	1✔
69	async with httpx.AsyncClient(verify=get_ssl_context(), timeout=5) as client:	1✔
70	resp = await client.post(self.gitlab_graphql_url, json=body, headers=header, timeout=10)	1✔
71	if resp.status_code != 200:	1✔
72	raise errors.BaseError(message=f"Error querying Gitlab api {self.gitlab_graphql_url}: {resp.text}")	×
73	result = cast(dict[str, Any], resp.json())	1✔
74
75	if "data" not in result or "projects" not in result["data"]:	1✔
76	raise errors.BaseError(message=f"Got unexpected response from Gitlab: {result}")	×
77	return result	1✔
78
79	resp_body = await _query_gitlab_graphql(body, header)	1✔
80	result: list[str] = []	1✔
81
82	def _process_projects(	1✔
83	resp_body: dict[str, Any], min_access_level: GitlabAccessLevel, result: list[str]
84	) -> None:
85	for project in resp_body["data"]["projects"]["nodes"]:	1✔
86	if min_access_level != GitlabAccessLevel.PUBLIC:	1✔
87	if not project["projectMembers"]["nodes"]:	1✔
88	continue	1✔
89	if min_access_level == GitlabAccessLevel.ADMIN:	1✔
90	max_level = max(	1✔
91	n["accessLevel"]["integerValue"]
92	for n in project["projectMembers"]["nodes"]
93	if n["user"]["id"].rsplit("/", maxsplit=1)[-1] == user.id
94	)
95	if max_level < 30:	1✔
96	continue	1✔
97	result.append(project["id"].rsplit("/", maxsplit=1)[-1])	1✔
98
99	_process_projects(resp_body, min_access_level, result)	1✔
100	page_info = resp_body["data"]["projects"]["pageInfo"]	1✔
101	while page_info["hasNextPage"]:	1✔
102	cursor = page_info["endCursor"]	×
103	body = {	×
104	"query": f"""{{
105	projects(ids: [{ids}], after: "{cursor}") {{
106	{query_body}
107	}}
108	}}
109	"""
110	}
111	resp_body = await _query_gitlab_graphql(body, header)	×
112	page_info = resp_body["data"]["projects"]["pageInfo"]	×
113	_process_projects(resp_body, min_access_level, result)	×
114
115	return result	1✔
116
117
118	@dataclass(kw_only=True)	2✔
119	class EmptyGitlabAPI:	2✔
120	"""An empty gitlab API used to decouple gitlab from Renku."""
121
122	async def filter_projects_by_access_level(self, _: APIUser, __: list[str], ___: GitlabAccessLevel) -> list[str]:	2✔
123	"""Always return an empty list."""
NEW 124	return []	×
125
126
127	@dataclass(kw_only=True)	2✔
128	class DummyGitlabAPI:	2✔
129	"""Dummy gitlab API.
130
131	The user with name Admin Doe has admin access to project 123456 and member access to 999999.
132	"""
133
134	_store = {	2✔
135	"Admin Doe": {
136	GitlabAccessLevel.MEMBER: ["999999", "123456"],
137	GitlabAccessLevel.ADMIN: ["123456"],
138	},
139	}
140
141	async def filter_projects_by_access_level(	2✔
142	self, user: APIUser, project_ids: list[str], min_access_level: GitlabAccessLevel
143	) -> list[str]:
144	"""Filter projects this user can access in gitlab with at least access level."""
145	if not user.access_token or not user.full_name:	2✔
146	return []	1✔
147	if min_access_level == GitlabAccessLevel.PUBLIC:	2✔
148	return []	×
149	user_projects = self._store.get(user.full_name, {}).get(min_access_level, [])	2✔
150	return [p for p in project_ids if p in user_projects]	2✔

SwissDataScienceCenter / renku-data-services / 17404558887

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