85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
import logging
|
|
from json import JSONDecodeError
|
|
from urllib.parse import urljoin
|
|
|
|
import requests
|
|
from pydantic import TypeAdapter, ValidationError
|
|
|
|
from .log_fields import OPERATION, PAGE, REPOSITORY_COUNT, STATUS_CODE
|
|
from .models import GiteaRepository
|
|
|
|
REPOSITORIES_ADAPTER = TypeAdapter(list[GiteaRepository])
|
|
logger = logging.getLogger(__name__)
|
|
REQUEST_TIMEOUT = (5, 30)
|
|
|
|
|
|
class GiteaApiError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class GiteaApi:
|
|
def __init__(self, endpoint: str, api_token: str):
|
|
self._endpoint = str(endpoint)
|
|
self._api_token = api_token
|
|
|
|
def get_repositories(self, page_size=10) -> list[GiteaRepository]:
|
|
"""
|
|
For mirroring input user is not important.
|
|
"""
|
|
session = requests.session()
|
|
session.headers.update({"Authorization": "token " + self._api_token})
|
|
all_repos = {} # hack for unique repositories in result
|
|
page_id = 1
|
|
while True:
|
|
try:
|
|
r = session.get(
|
|
urljoin(
|
|
self._endpoint,
|
|
"/api/v1/user/repos",
|
|
),
|
|
params={"limit": page_size, "page": page_id},
|
|
timeout=REQUEST_TIMEOUT,
|
|
)
|
|
except requests.RequestException as err:
|
|
raise GiteaApiError(
|
|
f"request failed while fetching page {page_id}: {err}"
|
|
) from err
|
|
if r.status_code != 200:
|
|
logger.error(
|
|
"failed to fetch repositories",
|
|
extra={
|
|
OPERATION: "fetch_repositories",
|
|
STATUS_CODE: r.status_code,
|
|
PAGE: page_id,
|
|
},
|
|
)
|
|
raise GiteaApiError(
|
|
f"failed to fetch repositories: status {r.status_code}"
|
|
)
|
|
try:
|
|
repos_data = r.json()
|
|
except (JSONDecodeError, ValueError) as err:
|
|
raise GiteaApiError(
|
|
f"invalid JSON while fetching page {page_id}: {err}"
|
|
) from err
|
|
if not repos_data:
|
|
break
|
|
else:
|
|
page_id += 1
|
|
try:
|
|
cur_repos = REPOSITORIES_ADAPTER.validate_python(repos_data)
|
|
except ValidationError as err:
|
|
raise GiteaApiError(
|
|
f"invalid repository payload on page {page_id - 1}: {err}"
|
|
) from err
|
|
for repo in cur_repos:
|
|
all_repos[repo.repo_id] = repo
|
|
logger.info(
|
|
"repositories fetched",
|
|
extra={
|
|
OPERATION: "fetch_repositories",
|
|
REPOSITORY_COUNT: len(all_repos),
|
|
},
|
|
)
|
|
return list(all_repos.values())
|