100 lines
2.6 KiB
Python
100 lines
2.6 KiB
Python
from dataclasses import dataclass, field, fields
|
|
from functools import cached_property
|
|
from json import loads
|
|
from typing import Any
|
|
|
|
from pynvim import Nvim
|
|
|
|
from .requests import request
|
|
from .ui import Select
|
|
|
|
from .clickup import ClickupTask
|
|
from .env import EnvVar
|
|
|
|
|
|
@dataclass
|
|
class GitlabMergeRequest:
|
|
@property
|
|
def updateables(self) -> dict[str, Any]:
|
|
return {f.name: getattr(self, f.name, None) for f in fields(type(self)) if f.repr}
|
|
|
|
|
|
@dataclass(kw_only=True)
|
|
class GitlabIssue:
|
|
project: int | str = field(repr=False)
|
|
"""project id or url encoded path to project, e.g. `ope%2Fopepipeline`"""
|
|
|
|
title: str
|
|
assignee_id: int | None = None
|
|
description: str | None = None
|
|
labels: list[str] | None = None
|
|
milestone_id: int | None = None
|
|
|
|
@property
|
|
def updateables(self) -> dict[str, Any]:
|
|
return {f.name: getattr(self, f.name, None) for f in fields(type(self)) if f.repr}
|
|
|
|
|
|
@dataclass
|
|
class GitlabSession:
|
|
auth_key: str = field(
|
|
default_factory=EnvVar(
|
|
"GITLAB_AUTH",
|
|
"gitlab autho token is required to be set",
|
|
),
|
|
)
|
|
user_id: str = field(
|
|
default_factory=EnvVar(
|
|
"GITLAB_USER_ID",
|
|
"gitlab user id is required to be set",
|
|
),
|
|
)
|
|
|
|
base_url: str = "https://git.extoll.de/api/v4"
|
|
|
|
def _authed_request(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
data: dict[str, str] | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
) -> str:
|
|
return request(
|
|
url,
|
|
method,
|
|
data=data,
|
|
headers=(headers or {}) | {"PRIVATE_TOKEN": self.auth_key},
|
|
)
|
|
|
|
@cached_property
|
|
def projects(self) -> list[str]:
|
|
return [
|
|
p["path_with_namespace"]
|
|
for p in loads(self._authed_request("GET", self.base_url + "/projects"))
|
|
]
|
|
|
|
def new_issue(self, nvim: Nvim, clickup_task: ClickupTask) -> GitlabIssue:
|
|
issue_new = GitlabIssue(
|
|
project=Select(nvim, self.projects)(),
|
|
title=clickup_task.name,
|
|
assignee_id=int(self.user_id),
|
|
description=f"#{clickup_task.id}",
|
|
)
|
|
self._authed_request(
|
|
"POST",
|
|
self.base_url + f"/projects/{issue_new.project}/issues",
|
|
data=issue_new.updateables,
|
|
)
|
|
|
|
return issue_new
|
|
|
|
def new_mr(self, from_issue: GitlabIssue) -> GitlabMergeRequest:
|
|
mr_new = GitlabMergeRequest()
|
|
self._authed_request(
|
|
"POST",
|
|
self.base_url + f"/projects/{from_issue.project}/merge_requests",
|
|
data=mr_new.updateables,
|
|
)
|
|
|
|
return mr_new
|