Patrick Nisble
134 lines
3.3 KiB
Python
134 lines
3.3 KiB
Python
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field
|
|
from requests import get
|
|
from requests.compat import quote
|
|
|
|
from eta.common import EnvVar, JSONDataMap
|
|
|
|
|
|
class GitlabProject(BaseModel):
|
|
name: str
|
|
name_with_namespace: str
|
|
path_with_namespace: str
|
|
web_url: str
|
|
id: int
|
|
|
|
issues: list["GitlabIssue"] = Field(default_factory=list, init=False)
|
|
merge_requests: list["GitlabMergeRequest"] = Field(default_factory=list, init=False)
|
|
|
|
|
|
class GitlabMilestone(BaseModel):
|
|
id: int
|
|
title: str
|
|
state: str
|
|
|
|
|
|
class GitlabIssue(BaseModel):
|
|
title: str
|
|
description: str | None
|
|
id: int
|
|
project_id: int
|
|
iid: int
|
|
state: str
|
|
labels: list[str]
|
|
milestone: GitlabMilestone | None
|
|
# web_url: str
|
|
|
|
|
|
class GitlabMergeRequest(BaseModel):
|
|
id: int
|
|
iid: int
|
|
description: str | None
|
|
draft: bool
|
|
labels: list[str]
|
|
milestone: GitlabMilestone | None
|
|
source_branch: str
|
|
state: str
|
|
target_branch: str
|
|
project_id: int
|
|
title: str
|
|
web_url: str
|
|
|
|
|
|
@dataclass
|
|
class GitlabSession:
|
|
auth_key: str = field(
|
|
default_factory=EnvVar(
|
|
"GITLAB_AUTH",
|
|
"gitlab auth token is required to be set",
|
|
)
|
|
)
|
|
base_url: str = "https://git.extoll.de/api/v4"
|
|
|
|
def _get(self, endpoint: str, **query_params: str) -> JSONDataMap:
|
|
with get(
|
|
self.base_url + endpoint,
|
|
query_params,
|
|
headers={
|
|
"accept": "application/json",
|
|
"PRIVATE-TOKEN": self.auth_key,
|
|
},
|
|
verify=False,
|
|
) as resp:
|
|
return resp.json()
|
|
|
|
def get_projects(self, **filters: str) -> list[GitlabProject]:
|
|
ret: list[GitlabProject] = []
|
|
page = 1
|
|
while projs := self._get(
|
|
"/projects",
|
|
active="true",
|
|
simple="true",
|
|
per_page="100",
|
|
page=str(page),
|
|
):
|
|
ret.extend(GitlabProject.model_validate(proj) for proj in projs)
|
|
page = page + 1
|
|
if len(projs) < 100:
|
|
break
|
|
|
|
return ret
|
|
|
|
def get_issues(self, projects: list[GitlabProject]) -> list[GitlabIssue]:
|
|
ret: list[GitlabIssue] = []
|
|
pids = {p.id for p in projects}
|
|
page = 1
|
|
while True:
|
|
issues = self._get(
|
|
"/issues",
|
|
per_page="100",
|
|
page=str(page),
|
|
)
|
|
ret.extend(
|
|
GitlabIssue.model_validate(iss) for iss in issues if int(iss["project_id"]) in pids
|
|
)
|
|
page = page + 1
|
|
if len(issues) < 100:
|
|
break
|
|
|
|
return ret
|
|
|
|
def get_merge_requests(self, projects: list[GitlabProject]) -> list[GitlabMergeRequest]:
|
|
ret: list[GitlabMergeRequest] = []
|
|
pids = {p.id for p in projects}
|
|
page = 1
|
|
while True:
|
|
merge_requests = self._get(
|
|
"/merge_requests",
|
|
simple="true",
|
|
per_page="100",
|
|
page=str(page),
|
|
)
|
|
ret.extend(
|
|
GitlabMergeRequest.model_validate(mr)
|
|
for mr in merge_requests
|
|
if int(mr["project_id"]) in pids
|
|
)
|
|
page = page + 1
|
|
if len(merge_requests) < 100:
|
|
break
|
|
|
|
return ret
|