171 lines
4.9 KiB
Python
171 lines
4.9 KiB
Python
"""
|
|
Vhex core backend for Gitea instance
|
|
"""
|
|
import os
|
|
import subprocess
|
|
import requests
|
|
|
|
from core.pkg.backend.core import VxRemoteBackendCore
|
|
from core.pkg.version import VxVersion
|
|
from core.logger import log
|
|
from core.config import config_get
|
|
|
|
|
|
__all__ = [
|
|
'VxBackendRemote'
|
|
]
|
|
|
|
class VxBackendRemote(VxRemoteBackendCore):
|
|
"""
|
|
Vhex Gitea backend class
|
|
"""
|
|
|
|
#---
|
|
# Public properties
|
|
#---
|
|
|
|
@property
|
|
def name(self):
|
|
return 'gitea'
|
|
|
|
@property
|
|
def url(self):
|
|
return self._url
|
|
|
|
@property
|
|
def package_list(self):
|
|
""" Return the list of all repository find in the remote instance.
|
|
|
|
This property is used to cache request's information and avoid big HTTP
|
|
quesring to the remote instance of Gitea.
|
|
"""
|
|
# if cached, avoid quering operation
|
|
if self._pkg_list:
|
|
return self._pkg_list
|
|
|
|
# raw HTTP request
|
|
resp = requests.get(
|
|
f'{self.url}/api/v1/repos/search',
|
|
params={
|
|
'q' : 'vxsdk',
|
|
'topic': 'True'
|
|
}
|
|
)
|
|
if not resp.ok:
|
|
log.warn(
|
|
f'[backend] gitea: remote package requests failed\n'
|
|
f'>>> {resp.status_code}'
|
|
)
|
|
self._pkg_list = []
|
|
return []
|
|
|
|
# generate the pacakge list
|
|
# @note
|
|
# - hide private repositories
|
|
self._pkg_list = []
|
|
for pkg in resp.json()['data']:
|
|
if pkg['private']:
|
|
continue
|
|
self._pkg_list.append({
|
|
'name' : pkg['name'],
|
|
'full_name' : pkg['full_name'],
|
|
'description' : pkg['description'],
|
|
'url' : pkg['clone_url'],
|
|
'created' : pkg['created_at'].split('T')[0],
|
|
'updated' : pkg['updated_at'].split('T')[0],
|
|
'author' : pkg['owner']['login'],
|
|
'default_branch' : pkg['default_branch'],
|
|
'versions' : []
|
|
})
|
|
return self._pkg_list
|
|
|
|
#---
|
|
# Private methods
|
|
#---
|
|
|
|
def package_fetch_versions(self, pkg):
|
|
""" Fetch package version information.
|
|
|
|
Same as the `package_list` property, we using cache to avoid too many
|
|
HTTP request.
|
|
"""
|
|
# if cached information, return it
|
|
if pkg['versions']:
|
|
return pkg['versions']
|
|
|
|
# request branches information
|
|
pkg['versions'] = []
|
|
url = f"{self.url}/api/v1/repos/{pkg['full_name']}"
|
|
resp = requests.get(f'{url}/branches')
|
|
if not resp.ok:
|
|
log.warn(
|
|
f'[pkg]: backend: gitea: branches requests error\n'
|
|
f'>>> url = {url}/branches\n'
|
|
f'>>> status = {resp.status_code}'
|
|
)
|
|
else:
|
|
for ver in resp.json():
|
|
pkg['versions'].append(
|
|
VxVersion(ver['name'], 'branch', 'remote')
|
|
)
|
|
|
|
# request tag information
|
|
resp = requests.get(f'{url}/tags')
|
|
if not resp.ok:
|
|
log.warn(
|
|
f'[pkg]: backend: gitea: tags requests error\n'
|
|
f'>>> url = {url}/tags\n'
|
|
f'>>> status = {resp.status_code}'
|
|
)
|
|
else:
|
|
for ver in resp.json():
|
|
pkg['versions'].append(VxVersion(ver['name'], 'tag', 'remote'))
|
|
return pkg['versions']
|
|
|
|
#---
|
|
# Public methods
|
|
#---
|
|
|
|
def package_clone(self, pkg, _=None):
|
|
""" Clone the package in global storage
|
|
|
|
@args
|
|
> pkg (dict) - package information returned by package_find()
|
|
|
|
@return
|
|
> Complet path for the package (str), or None if error
|
|
"""
|
|
# fetch global storage prefix
|
|
# @notes
|
|
# - create it if its doesn't exists
|
|
prefix = os.path.expanduser(config_get('path.packages'))
|
|
if not os.path.exists(prefix):
|
|
os.makedirs(prefix)
|
|
|
|
# generate clone information
|
|
# @note
|
|
# - create clone folder if not exists
|
|
pkg_ver = pkg['version']
|
|
pkg_name = f"{pkg['author']}@{pkg['name']}@{pkg_ver.name}@{pkg_ver.type}"
|
|
pkg_path = f"{prefix}/{pkg_name}"
|
|
if os.path.exists(pkg_path):
|
|
log.warn(f"[clone]: {pkg_name} already exists, skipped")
|
|
return pkg_path
|
|
|
|
# perform high-level git clone
|
|
cmd = [
|
|
'git',
|
|
'-c', 'advice.detachedHead=false',
|
|
'clone', '--branch', pkg_ver.name,
|
|
pkg['url'], pkg_path,
|
|
'--depth=1'
|
|
]
|
|
log.debug(f"[gitea] {cmd}")
|
|
status = subprocess.run(cmd, capture_output=True, check=False)
|
|
if status.returncode != 0:
|
|
log.error(f"[clone] : unable to clone {pkg_name}, abord")
|
|
return []
|
|
|
|
# return the package path
|
|
return pkg_path
|