119 lines
4.6 KiB
Python
119 lines
4.6 KiB
Python
import aiohttp
|
|
|
|
from ..logger import logger
|
|
from ..models import Issue, Repository
|
|
|
|
|
|
def _populate_headers(repository: Repository) -> dict[str, str]:
|
|
headers: dict[str, str] = {"User-Agent": repository.user_agent}
|
|
if repository.provider.token is not None:
|
|
headers["Authorization"] = f"Bearer {repository.provider.token}"
|
|
else:
|
|
logger.debug("GitLab API key is not set for provider '%s'. Using unauthenticated request.", repository.provider.url)
|
|
return headers
|
|
|
|
|
|
async def fetch_gitlab_issue(repository: Repository, issue_number: int) -> Issue:
|
|
"""
|
|
Fetch an issue from a GitLab instance.
|
|
Args:
|
|
repository (Repository): The repository to fetch the issue from.
|
|
issue_number (int): The issue number.
|
|
Returns:
|
|
Issue: An Issue object containing the issue details.
|
|
Raises:
|
|
aiohttp.ClientResponseError: If the request to the Forgejo API fails.
|
|
ValueError: If the response from the Forgejo API is empty
|
|
"""
|
|
url = f"{repository.provider.api_url}/projects/{repository.owner}%2F{repository.name}/issues/{issue_number}"
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url, headers=_populate_headers(repository)) as response:
|
|
if response.status != 200:
|
|
raise aiohttp.ClientResponseError(
|
|
request_info=response.request_info,
|
|
history=response.history,
|
|
status=response.status,
|
|
message=f"Failed to fetch issue '{issue_number}' from '{repository.url}'.",
|
|
)
|
|
data = await response.json()
|
|
|
|
if data is None:
|
|
raise ValueError("Received empty response from the GitLab API.")
|
|
|
|
if data.get("state", "opened") == "closed":
|
|
issue_type = "issue_closed"
|
|
else:
|
|
issue_type = "issue"
|
|
|
|
return Issue(
|
|
repository=repository,
|
|
number=data["iid"],
|
|
author=data["author"]["username"],
|
|
author_avatar=data["author"]["avatar_url"],
|
|
author_url=data["author"]["web_url"],
|
|
link=data["web_url"],
|
|
title=data["title"],
|
|
body=data["description"],
|
|
type=issue_type,
|
|
labels=data.get("labels", []),
|
|
draft=False,
|
|
creation_date=data["created_at"],
|
|
milestone=data.get("milestone", {}).get("title") if data.get("milestone") is not None else None,
|
|
response=response,
|
|
)
|
|
|
|
|
|
async def fetch_gitlab_merge_request(repository: Repository, merge_request_number: int) -> Issue:
|
|
"""
|
|
Fetch a merge request from a GitLab instance.
|
|
Args:
|
|
repository (Repository): The repository to fetch the merge request from.
|
|
merge_request_number (int): The merge request number.
|
|
Returns:
|
|
Issue: An Issue object containing the merge request details.
|
|
Raises:
|
|
aiohttp.ClientResponseError: If the request to the GitLab API fails.
|
|
ValueError: If the response from the GitLab API is empty
|
|
"""
|
|
url = f"{repository.provider.api_url}/projects/{repository.owner}%2F{repository.name}/merge_requests/{merge_request_number}"
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url, headers=_populate_headers(repository)) as response:
|
|
if response.status != 200:
|
|
raise aiohttp.ClientResponseError(
|
|
request_info=response.request_info,
|
|
history=response.history,
|
|
status=response.status,
|
|
message=f"Failed to fetch merge request '{merge_request_number}' from '{repository.url}'.",
|
|
)
|
|
data = await response.json()
|
|
|
|
if data is None:
|
|
raise ValueError("Received empty response from the GitLab API.")
|
|
|
|
state = data.get("state", "opened")
|
|
if state == "merged":
|
|
issue_type = "pull_request_merged"
|
|
elif state == "closed":
|
|
issue_type = "pull_request_closed"
|
|
elif data.get("draft", False) is True:
|
|
issue_type = "pull_request_draft"
|
|
else:
|
|
issue_type = "pull_request"
|
|
|
|
return Issue(
|
|
repository=repository,
|
|
number=data["iid"],
|
|
author=data["author"]["username"],
|
|
author_avatar=data["author"]["avatar_url"],
|
|
author_url=data["author"]["web_url"],
|
|
link=data["web_url"],
|
|
title=data["title"],
|
|
body=data["description"],
|
|
type=issue_type,
|
|
labels=data.get("labels", []),
|
|
draft=data.get("draft", False),
|
|
creation_date=data["created_at"],
|
|
milestone=data.get("milestone", {}).get("title") if data.get("milestone") is not None else None,
|
|
merge_date=data.get("merged_at"),
|
|
response=response,
|
|
)
|