initial: extract altrepo module as standalone library

parents
__pycache__/
*.py[cod]
*.egg-info/
dist/
build/
.venv/
import aiohttp
from .api import ALTRepoAPI
from .appstream import ALTRepoAppStream
from .config import ALTRepoConfig
from .parser import ALTRepoParser
class ALTRepo:
def __init__(self, config: ALTRepoConfig | None = None):
self.config = config or ALTRepoConfig()
self._session: aiohttp.ClientSession | None = None
self.api: ALTRepoAPI | None = None
self.appstream: ALTRepoAppStream | None = None
self.parser: ALTRepoParser | None = None
async def init(self, session: aiohttp.ClientSession | None = None):
if self._session is None:
self._session = session or aiohttp.ClientSession()
self.api = ALTRepoAPI(self._session, self.config)
self.appstream = ALTRepoAppStream(self._session, self.config)
self.parser = ALTRepoParser(self._session, self.config)
async def close(self):
await self._session.close()
__all__ = ("ALTRepo", "ALTRepoConfig")
from .methods import ALTRepoAPI
class RequestValidationError(Exception):
"""Ошибка валидации параметров запроса (400)"""
pass
class DataNotFoundError(Exception):
"""Данные не найдены в базе (404)"""
pass
class TooManyRequests(Exception):
"""Cлишком много запросо (429)"""
pass
class UnexpectedResponseError(Exception):
"""Непредвиденный ответ от сервера"""
def __init__(self, status_code: int, message: str = ""):
self.status_code = status_code
self.message = message or f"Unexpected response status: {status_code}"
super().__init__(self.message)
import aiohttp
from urllib.parse import urlencode
from typing import List, Literal
from . import models
from . import errors
class BaseAPI:
def __init__(self, session: aiohttp.ClientSession, base_url: str):
self.session = session
self.BASE_URL = base_url
async def _handle_response(self, resp: aiohttp.ClientResponse):
match resp.status:
case 200 | 201:
return await resp.json()
case 400:
raise errors.RequestValidationError(
"Request parameters validation error (400)"
)
case 404:
raise errors.DataNotFoundError(
"Requested data not found in database (404)"
)
case 429:
raise errors.TooManyRequests("Too Many Requests (429)")
case _:
try:
error_text = await resp.text()
except:
error_text = ""
raise errors.UnexpectedResponseError(resp.status, error_text)
async def get(self, path: str, params: dict = None):
url = f"{self.BASE_URL}/{path.lstrip('/')}"
if params:
query = urlencode(
{
k: ",".join(v) if isinstance(v, list) else v
for k, v in params.items()
}
)
url += f"?{query}"
async with self.session.get(url) as resp:
return await self._handle_response(resp)
async def post(self, path: str, data: dict = None, json: dict = None):
url = f"{self.BASE_URL}/{path.lstrip('/')}"
async with self.session.post(url, data=data, json=json) as resp:
return await self._handle_response(resp)
async def get_text(self, path: str, params: dict = None) -> str:
url = f"{self.BASE_URL}/{path.lstrip('/')}"
if params:
query = urlencode(
{
k: ",".join(v) if isinstance(v, list) else v
for k, v in params.items()
}
)
url += f"?{query}"
async with self.session.get(url) as resp:
if resp.status == 200:
return await resp.text()
await self._handle_response(resp)
async def get_bytes(self, path: str, params: dict = None) -> bytes:
url = f"{self.BASE_URL}/{path.lstrip('/')}"
if params:
query = urlencode(
{
k: ",".join(v) if isinstance(v, list) else v
for k, v in params.items()
}
)
url += f"?{query}"
async with self.session.get(url) as resp:
if resp.status == 200:
return await resp.read()
await self._handle_response(resp)
class APIInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def version(self) -> models.APIVersion:
data = await self.client.get("/version")
return models.APIVersion(**data)
class TaskProcessInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def find_tasks(
self,
input: list[str],
owner: str | None = None,
branch: str | None = None,
state: list[str] | None = None,
tasks_limit: int = 100,
by_package: bool = False,
) -> models.TasksListModel:
params = {
k: v
for k, v in {
"input": input,
"owner": owner,
"branch": branch,
"state": state,
"tasks_limit": tasks_limit,
"by_package": by_package,
}.items()
if v is not None
}
data = await self.client.get("/task/progress/find_tasks", params)
return models.TasksListModel(**data)
async def task_info(self, task_id: int) -> models.TaskProgressTaskInfoModel:
data = await self.client.get(f"/task/progress/task_info/{task_id}")
return models.TaskProgressTaskInfoModel(**data)
async def all_packagesets(self) -> models.AllTasksBranchesModel:
data = await self.client.get("/task/progress/all_packagesets")
return models.AllTasksBranchesModel(**data)
async def all_tasks_branches(self) -> models.AllTasksBranchesModel:
data = await self.client.get("/task/progress/all_tasks_branches")
return models.AllTasksBranchesModel(**data)
async def find_tasks_lookup(
self,
input: list[str],
branch: str | None = None,
tasks_limit: int = 10,
) -> models.FindTasksModel:
params = {
k: v
for k, v in {
"input": input,
"branch": branch,
"tasks_limit": tasks_limit,
}.items()
if v is not None
}
data = await self.client.get("/task/progress/find_tasks_lookup", params)
return models.FindTasksModel(**data)
async def last_tasks(
self, branch: str | None = None, tasks_limit: int = 10
) -> models.TasksListModel:
params = {
k: v
for k, v in {
"branch": branch,
"tasks_limit": tasks_limit,
}.items()
if v is not None
}
data = await self.client.get("/task/progress/last_tasks", params)
return models.TasksListModel(**data)
class TaskInfo:
def __init__(self, client: BaseAPI):
self.client = client
self.progress = TaskProcessInfo(self.client)
async def build_dependency_set(
self, task_id: int, arch: str = "x86_64"
) -> models.BuildDependencySetModel:
data = await self.client.get(
f"/task/build_dependency_set/{task_id}", {"arch": arch}
)
return models.BuildDependencySetModel(**data)
async def check_images(self, payload: dict) -> models.CheckImagesOutputModel:
data = await self.client.post("/task/check_images", json=payload)
return models.CheckImagesOutputModel(**data)
async def find_images(self, task_id: int) -> models.FindImagesByTaskModel:
data = await self.client.get(f"/task/find_images/{task_id}")
return models.FindImagesByTaskModel(**data)
async def find_packageset(
self, task_id: int, branches: list[str] | None = None
) -> models.TaskFindPackagesetModel:
params = {
k: v
for k, v in {
"branches": branches,
}.items()
if v is not None
}
data = await self.client.get(f"/task/find_packageset/{task_id}", params or None)
return models.TaskFindPackagesetModel(**data)
async def misconflict(
self, task_id: int, archs: list[str] | None = None, no_cache: str | None = None
) -> models.TaskMisconflictPackagesModel:
params = {
k: v
for k, v in {
"archs": archs,
"no_cache": no_cache,
}.items()
if v is not None
}
data = await self.client.get(f"/task/misconflict/{task_id}", params or None)
return models.TaskMisconflictPackagesModel(**data)
async def needs_approval(
self,
acl_group: Literal["maint", "tester"],
branches: list[str] | None = None,
before: str | None = None,
) -> models.NeedsApprovalModel:
params = {
k: v
for k, v in {
"acl_group": acl_group,
"branches": branches,
"before": before,
}.items()
if v is not None
}
data = await self.client.get("/task/needs_approval", params)
return models.NeedsApprovalModel(**data)
async def packages(self, task_id: int) -> models.TaskPackagesModel:
data = await self.client.get(f"/task/packages/{task_id}")
return models.TaskPackagesModel(**data)
async def task_diff(self, task_id: int) -> models.TaskDiffModel:
data = await self.client.get(f"/task/task_diff/{task_id}")
return models.TaskDiffModel(**data)
async def task_history(
self,
branch: str,
start_task: int = 0,
end_task: int = 0,
start_date: str | None = None,
end_date: str | None = None,
) -> models.TaskHistoryModel:
params = {
k: v
for k, v in {
"branch": branch,
"start_task": start_task,
"end_task": end_task,
"start_date": start_date,
"end_date": end_date,
}.items()
if v is not None
}
data = await self.client.get("/task/task_history", params)
return models.TaskHistoryModel(**data)
async def task_info(
self,
task_id: int,
try_: int | None = None,
iteration: int | None = None,
no_cache: str | None = None,
states: list[str] | None = None,
) -> models.TaskInfoModel:
params = {
k: v
for k, v in {
"try": try_,
"iteration": iteration,
"no_cache": no_cache,
"states": states,
}.items()
if v is not None
}
data = await self.client.get(f"/task/task_info/{task_id}", params or None)
return models.TaskInfoModel(**data)
async def task_repo(
self, task_id: int, include_task_packages: bool = False
) -> models.TaskRepoModel:
data = await self.client.get(
f"/task/task_repo/{task_id}",
{"include_task_packages": include_task_packages},
)
return models.TaskRepoModel(**data)
async def what_depends_src(
self,
task_id: int,
depth: int = 1,
dptype: Literal["both", "source", "binary"] = "both",
archs: list[str] | None = None,
leaf: str | None = None,
finite_package: bool = False,
filter_by_package: list[str] | None = None,
filter_by_source: str | None = None,
oneandhalf: bool = False,
no_cache: str | None = None,
) -> models.TaskBuildDependencyModel:
params = {
k: v
for k, v in {
"depth": depth,
"dptype": dptype,
"archs": archs,
"leaf": leaf,
"finite_package": finite_package,
"filter_by_package": filter_by_package,
"filter_by_source": filter_by_source,
"oneandhalf": oneandhalf,
"no_cache": no_cache,
}.items()
if v is not None
}
data = await self.client.get(
f"/task/what_depends_src/{task_id}", params or None
)
return models.TaskBuildDependencyModel(**data)
class PackageInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def package_info(
self,
name: str | None = None,
version: str | None = None,
release: str | None = None,
arch: str | None = None,
source: bool | None = None,
branch: str | None = None,
disttag: str | None = None,
sha1: str | None = None,
packager: str | None = None,
packager_email: str | None = None,
full: bool | None = None,
) -> models.PackageInfoModel:
params = {
k: v
for k, v in {
"name": name,
"version": version,
"release": release,
"arch": arch,
"source": source,
"branch": branch,
"disttag": disttag,
"sha1": sha1,
"packager": packager,
"packager_email": packager_email,
"full": full,
}.items()
if v is not None
}
data = await self.client.get("/package/package_info", params)
return models.PackageInfoModel(**data)
async def packages_by_file_names(
self, files: List[str], branch: str, arch: str | None = None
) -> models.PackageByFileNameModel:
payload = models.PackagesByFileNamesJsonModel(
files=files, branch=branch, arch=arch
)
data = await self.client.post(
"/package/packages_by_file_names",
json=payload.model_dump(exclude_none=True),
)
return models.PackageByFileNameModel(**data)
async def build_dependency_set(
self,
branch: str,
packages: list[str],
arch: str = "x86_64",
) -> models.BuildDependencySetModel:
data = await self.client.get(
"/package/build_dependency_set",
{"branch": branch, "packages": packages, "arch": arch},
)
return models.BuildDependencySetModel(**data)
async def find_packageset(
self, packages: list[str], branches: list[str] | None = None
) -> models.PackageFindPackagesetModel:
params = {
k: v
for k, v in {
"packages": packages,
"branches": branches,
}.items()
if v is not None
}
data = await self.client.get("/package/find_packageset", params)
return models.PackageFindPackagesetModel(**data)
async def maintainer_score(
self, branch: str, name: str
) -> models.MaintainerScoreModel:
data = await self.client.get(
"/package/maintainer_score", {"branch": branch, "name": name}
)
return models.MaintainerScoreModel(**data)
async def misconflict(
self,
packages: list[str],
branch: str,
archs: list[str] | None = None,
) -> models.PackageMisconflictPackagesModel:
params = {
k: v
for k, v in {
"packages": packages,
"branch": branch,
"archs": archs,
}.items()
if v is not None
}
data = await self.client.get("/package/misconflict", params)
return models.PackageMisconflictPackagesModel(**data)
async def package_by_file_md5(
self, branch: str, md5: str, arch: str = "x86_64"
) -> models.PackageByFileNameModel:
data = await self.client.get(
"/package/package_by_file_md5", {"branch": branch, "md5": md5, "arch": arch}
)
return models.PackageByFileNameModel(**data)
async def package_by_file_name(
self, file: str, branch: str, arch: str = "x86_64"
) -> models.PackageByFileNameModel:
data = await self.client.get(
"/package/package_by_file_name",
{"file": file, "branch": branch, "arch": arch},
)
return models.PackageByFileNameModel(**data)
async def package_files(self, pkghash: int) -> models.PackageFilesModel:
data = await self.client.get(f"/package/package_files/{pkghash}")
return models.PackageFilesModel(**data)
async def repocop(
self,
branch: str,
package_name: str,
package_version: str | None = None,
package_release: str | None = None,
bin_package_arch: str | None = None,
package_type: Literal["source", "binary"] = "source",
) -> models.RepocopJsonGetListModel:
params = {
k: v
for k, v in {
"branch": branch,
"package_name": package_name,
"package_version": package_version,
"package_release": package_release,
"bin_package_arch": bin_package_arch,
"package_type": package_type,
}.items()
if v is not None
}
data = await self.client.get("/package/repocop", params)
return models.RepocopJsonGetListModel(**data)
async def repocop_post(self, packages: list[dict]) -> None:
await self.client.post("/package/repocop", json={"packages": packages})
async def specfile_by_hash(self, pkghash: int) -> models.PackageSpecfileModel:
data = await self.client.get(f"/package/specfile_by_hash/{pkghash}")
return models.PackageSpecfileModel(**data)
async def specfile_by_name(
self, branch: str, name: str
) -> models.PackageSpecfileModel:
data = await self.client.get(
"/package/specfile_by_name", {"branch": branch, "name": name}
)
return models.PackageSpecfileModel(**data)
async def unpackaged_dirs(
self, branch: str, packager: str, archs: list[str] | None = None
) -> models.UnpackagedDirsModel:
params = {
k: v
for k, v in {
"branch": branch,
"packager": packager,
"archs": archs,
}.items()
if v is not None
}
data = await self.client.get("/package/unpackaged_dirs", params)
return models.UnpackagedDirsModel(**data)
async def what_depends_src(
self,
packages: list[str],
branch: str,
depth: int = 1,
dptype: Literal["both", "source", "binary"] = "both",
archs: list[str] | None = None,
leaf: str | None = None,
finite_package: bool = False,
filter_by_package: list[str] | None = None,
filter_by_source: str | None = None,
oneandhalf: bool = False,
use_last_tasks: bool = False,
) -> models.PackageBuildDependencyModel:
params = {
k: v
for k, v in {
"packages": packages,
"branch": branch,
"depth": depth,
"dptype": dptype,
"archs": archs,
"leaf": leaf,
"finite_package": finite_package,
"filter_by_package": filter_by_package,
"filter_by_source": filter_by_source,
"oneandhalf": oneandhalf,
"use_last_tasks": use_last_tasks,
}.items()
if v is not None
}
data = await self.client.get("/package/what_depends_src", params)
return models.PackageBuildDependencyModel(**data)
class PackagesetInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def active_packagesets(self) -> models.PackageSetActivePackageSetsModel:
data = await self.client.get("/packageset/active_packagesets")
return models.PackageSetActivePackageSetsModel(**data)
async def repository_statistics(
self, branch: str | None = None
) -> models.RepositoryStatisticsModel:
if branch:
data = await self.client.get(
"/packageset/repository_statistics", {"branch": branch}
)
else:
data = await self.client.get("/packageset")
return models.RepositoryStatisticsModel(**data)
async def compare_packagesets(
self, pkgset1: str, pkgset2: str
) -> models.PackagesetCompareModel:
data = await self.client.get(
"/packageset/compare_packagesets", {"pkgset1": pkgset1, "pkgset2": pkgset2}
)
return models.PackagesetCompareModel(**data)
async def maintainer_scores(self, branch: str) -> models.MaintainerScoresBatchModel:
data = await self.client.get(
"/packageset/maintainer_scores", {"branch": branch}
)
return models.MaintainerScoresBatchModel(**data)
async def packages_by_component(
self, branch: str, arch: str, component: str
) -> models.PackagesByUuidModel:
data = await self.client.get(
"/packageset/packages_by_component",
{"branch": branch, "arch": arch, "component": component},
)
return models.PackagesByUuidModel(**data)
async def packages_by_uuid(self, uuid: str) -> models.PackagesByUuidModel:
data = await self.client.get("/packageset/packages_by_uuid", {"uuid": uuid})
return models.PackagesByUuidModel(**data)
async def pkgset_status_get(self) -> models.PackageSetStatusGetModel:
data = await self.client.get("/packageset/pkgset_status")
return models.PackageSetStatusGetModel(**data)
async def pkgset_status_post(self, payload: dict) -> None:
await self.client.post("/packageset/pkgset_status", json=payload)
async def repository_packages(
self,
branch: str,
package_type: Literal["all", "source", "binary"] = "all",
archs: list[str] | None = None,
include_done_tasks: bool = False,
) -> models.PackagesetPackagesModel:
params = {
k: v
for k, v in {
"branch": branch,
"package_type": package_type,
"archs": archs,
"include_done_tasks": include_done_tasks,
}.items()
if v is not None
}
data = await self.client.get("/packageset/repository_packages", params)
return models.PackagesetPackagesModel(**data)
class AclInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def groups(self, branch: str, name: str | None = None):
data = {"branch": branch}
if name:
data["name"] = name
data = await self.client.get("/acl/groups", data)
return models.AclGroupsModel(**data)
async def by_packages(
self, branch: str, packages_names: list[str]
) -> models.AclByPackagesModel:
params = {
"branch": branch,
"packages_names": packages_names,
}
data = await self.client.get("/acl/by_packages", params)
return models.AclByPackagesModel(**data)
async def maintainer_groups(
self, nickname: str, branch: list[str] | None = None
) -> models.AclMaintainerGroupsModel:
params = {
k: v
for k, v in {
"nickname": nickname,
"branch": branch,
}.items()
if v is not None
}
data = await self.client.get("/acl/maintainer_groups", params)
return models.AclMaintainerGroupsModel(**data)
class BugInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def bugzilla_by_maintainer(
self, maintainer_nickname: str, by_acl: str | None = None
) -> models.BugzillaInfoModel | None:
try:
data = await self.client.get(
"/bug/bugzilla_by_maintainer",
{
"maintainer_nickname": maintainer_nickname,
"by_acl": by_acl if by_acl else "none",
},
)
except errors.DataNotFoundError:
return None
return models.BugzillaInfoModel(**data)
async def bugzilla_by_package(
self,
package_name: str,
package_type: Literal["source", "binary"] = "source",
) -> models.BugzillaInfoModel:
data = await self.client.get(
"/bug/bugzilla_by_package",
{"package_name": package_name, "package_type": package_type},
)
return models.BugzillaInfoModel(**data)
async def bugzilla_by_image_edition(
self, branch: str, edition: str
) -> models.BugzillaInfoModel:
data = await self.client.get(
"/bug/bugzilla_by_image_edition", {"branch": branch, "edition": edition}
)
return models.BugzillaInfoModel(**data)
class FileInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def packages_by_file(
self, branch: str, file_name: str
) -> models.FilePackagesByFileModel:
data = await self.client.get(
"/file/packages_by_file", {"branch": branch, "file_name": file_name}
)
return models.FilePackagesByFileModel(**data)
async def search(
self, branch: str, file_name: str, limit: int | None = None
) -> models.FilesModel:
data = await self.client.get(
"/file/search",
{
"branch": branch,
"file_name": file_name,
"limit": limit if limit else 1000,
},
)
return models.FilesModel(**data)
async def fast_lookup(
self, branch: str, file_name: str, limit: int = 10
) -> models.FastFileSearchModel:
data = await self.client.get(
"/file/fast_lookup",
{"branch": branch, "file_name": file_name, "limit": limit},
)
return models.FastFileSearchModel(**data)
class SiteInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def all_maintainers(self, branch: str) -> models.AllMaintainersModel:
return await self.all_maintainers_with_nicknames(branch)
async def all_maintainers_with_nicknames(
self, branch: str
) -> models.AllMaintainersModel:
data = await self.client.get(
"/site/all_maintainers_with_nicknames", {"branch": branch}
)
return models.AllMaintainersModel(**data)
async def find_source_package(
self, branch: str, name: str
) -> models.FindSourcePackageInBranch:
data = await self.client.get(
"/site/find_source_package", {"branch": branch, "name": name}
)
return models.FindSourcePackageInBranch(**data)
async def maintainer_info(
self, branch: str, maintainer_nickname: str
) -> models.MaintainerInfoModel:
data = await self.client.get(
"/site/maintainer_info",
{"branch": branch, "maintainer_nickname": maintainer_nickname},
)
return models.MaintainerInfoModel(**data)
async def package_info(
self,
branch: str,
pkghash: int,
changelog_last: int = 3,
package_type: Literal["source", "binary"] = "source",
) -> models.SitePackageInfoModel:
data = await self.client.get(
f"/site/package_info/{pkghash}",
{
"branch": branch,
"changelog_last": changelog_last,
"package_type": package_type,
},
)
return models.SitePackageInfoModel(**data)
async def pkghash_by_binary_name(
self, branch: str, name: str, arch: str
) -> models.SitePackagesetPackageHashModel:
data = await self.client.get(
"/site/pkghash_by_binary_name",
{"branch": branch, "name": name, "arch": arch},
)
return models.SitePackagesetPackageHashModel(**data)
async def pkghash_by_name(
self, branch: str, name: str
) -> models.SitePackagesetPackageHashModel:
data = await self.client.get(
"/site/pkghash_by_name", {"branch": branch, "name": name}
)
return models.SitePackagesetPackageHashModel(**data)
async def watch_by_maintainer(
self, maintainer_nickname: str
) -> models.SiteWatchByMaintainerModel | None:
try:
data = await self.client.get(
"/site/watch_by_maintainer",
{"maintainer_nickname": maintainer_nickname},
)
except errors.DataNotFoundError:
return None
return models.SiteWatchByMaintainerModel(**data)
async def all_pkgset_archs(self, branch: str) -> models.SiteAllArchsModel:
data = await self.client.get("/site/all_pkgset_archs", {"branch": branch})
return models.SiteAllArchsModel(**data)
async def all_pkgset_archs_with_src_count(
self, branch: str
) -> models.SiteAllArchsModel:
data = await self.client.get(
"/site/all_pkgset_archs_with_src_count", {"branch": branch}
)
return models.SiteAllArchsModel(**data)
async def all_pkgsets(self) -> models.SiteAllPackagasetsModel:
data = await self.client.get("/site/all_pkgsets")
return models.SiteAllPackagasetsModel(**data)
async def all_pkgsets_summary(self) -> models.SiteAllPackagesetsSummaryModel:
data = await self.client.get("/site/all_pkgsets_summary")
return models.SiteAllPackagesetsSummaryModel(**data)
async def all_pkgsets_with_src_count(self) -> models.SiteAllPackagasetsModel:
data = await self.client.get("/site/all_pkgsets_with_src_count")
return models.SiteAllPackagasetsModel(**data)
async def beehive_errors_by_maintainer(
self,
branch: str,
maintainer_nickname: str,
by_acl: str = "none",
) -> models.SiteBeehiveByMaintainerModel:
data = await self.client.get(
"/site/beehive_errors_by_maintainer",
{
"branch": branch,
"maintainer_nickname": maintainer_nickname,
"by_acl": by_acl,
},
)
return models.SiteBeehiveByMaintainerModel(**data)
async def binary_package_archs_and_versions(
self, branch: str, name: str
) -> models.SitePackagesBinaryListModel:
data = await self.client.get(
"/site/binary_package_archs_and_versions", {"branch": branch, "name": name}
)
return models.SitePackagesBinaryListModel(**data)
async def binary_package_scripts(
self, pkghash: int
) -> models.SiteBinPackageScriptsModel:
data = await self.client.get(f"/site/binary_package_scripts/{pkghash}")
return models.SiteBinPackageScriptsModel(**data)
async def deleted_package_info(
self,
branch: str,
name: str,
package_type: Literal["source", "binary"] = "source",
arch: str | None = None,
) -> models.SiteDeletedPackageModel:
params = {
k: v
for k, v in {
"branch": branch,
"name": name,
"package_type": package_type,
"arch": arch,
}.items()
if v is not None
}
data = await self.client.get("/site/deleted_package_info", params)
return models.SiteDeletedPackageModel(**data)
async def fast_packages_search_lookup(
self, name: list[str], branch: str | None = None
) -> models.SiteFastPackagesSearchModel:
params = {
k: v
for k, v in {
"name": name,
"branch": branch,
}.items()
if v is not None
}
data = await self.client.get("/site/fast_packages_search_lookup", params)
return models.SiteFastPackagesSearchModel(**data)
async def find_packages(
self,
name: list[str],
branch: str | None = None,
arch: list[str] | None = None,
) -> models.SiteFingPackagesModel:
params = {
k: v
for k, v in {
"name": name,
"branch": branch,
"arch": arch,
}.items()
if v is not None
}
data = await self.client.get("/site/find_packages", params)
return models.SiteFingPackagesModel(**data)
async def last_packages(
self,
branch: str,
tasks_limit: int = 10,
task_owner: str | None = None,
) -> models.SiteLastPackagesModel:
params = {
k: v
for k, v in {
"branch": branch,
"tasks_limit": tasks_limit,
"task_owner": task_owner,
}.items()
if v is not None
}
data = await self.client.get("/site/last_packages", params)
return models.SiteLastPackagesModel(**data)
async def last_packages_by_branch(
self,
branch: str,
packages_limit: int = 10,
packager: str | None = None,
) -> models.SiteLastBranchPackagesModel:
params = {
k: v
for k, v in {
"branch": branch,
"packages_limit": packages_limit,
"packager": packager,
}.items()
if v is not None
}
data = await self.client.get("/site/last_packages_by_branch", params)
return models.SiteLastBranchPackagesModel(**data)
async def last_packages_by_tasks(
self,
branch: str,
tasks_limit: int = 10,
task_owner: str | None = None,
) -> models.SiteLastPackagesModel:
params = {
k: v
for k, v in {
"branch": branch,
"tasks_limit": tasks_limit,
"task_owner": task_owner,
}.items()
if v is not None
}
data = await self.client.get("/site/last_packages_by_tasks", params)
return models.SiteLastPackagesModel(**data)
async def last_packages_with_cve_fixed(
self, branch: str
) -> models.SiteLastPackagesWithCVEFixesModel:
data = await self.client.get(
"/site/last_packages_with_cve_fixed", {"branch": branch}
)
return models.SiteLastPackagesWithCVEFixesModel(**data)
async def maintainer_branches(
self, maintainer_nickname: str
) -> models.MaintainerBranchesModel:
data = await self.client.get(
"/site/maintainer_branches", {"maintainer_nickname": maintainer_nickname}
)
return models.MaintainerBranchesModel(**data)
async def maintainer_packages(
self,
branch: str,
maintainer_nickname: str,
by_acl: str = "none",
) -> models.MaintainerPackagesModel:
data = await self.client.get(
"/site/maintainer_packages",
{
"branch": branch,
"maintainer_nickname": maintainer_nickname,
"by_acl": by_acl,
},
)
return models.MaintainerPackagesModel(**data)
async def package_changelog(
self, pkghash: int, changelog_last: int = 3
) -> models.SiteChangelogModel:
data = await self.client.get(
f"/site/package_changelog/{pkghash}", {"changelog_last": changelog_last}
)
return models.SiteChangelogModel(**data)
async def package_downloads(
self, pkghash: int, branch: str
) -> models.SitePackagesDownloadsModel:
data = await self.client.get(
f"/site/package_downloads/{pkghash}", {"branch": branch}
)
return models.SitePackagesDownloadsModel(**data)
async def package_downloads_bin(
self, pkghash: int, branch: str, arch: str
) -> models.SitePackagesDownloadsModel:
data = await self.client.get(
f"/site/package_downloads_bin/{pkghash}", {"branch": branch, "arch": arch}
)
return models.SitePackagesDownloadsModel(**data)
async def package_downloads_src(
self, pkghash: int, branch: str
) -> models.SitePackagesDownloadsModel:
data = await self.client.get(
f"/site/package_downloads_src/{pkghash}", {"branch": branch}
)
return models.SitePackagesDownloadsModel(**data)
async def package_info_brief(
self, pkghash: int
) -> models.SiteBriefPackageInfoModel:
data = await self.client.get(f"/site/package_info_brief/{pkghash}")
return models.SiteBriefPackageInfoModel(**data)
async def package_log_bin(self, pkghash: int) -> models.BinPackageLogElementModel:
data = await self.client.get(f"/site/package_log_bin/{pkghash}")
return models.BinPackageLogElementModel(**data)
async def package_misconflict(
self, pkghash: int, branch: str
) -> models.PackageMisconflictBySrcModel:
data = await self.client.get(
f"/site/package_misconflict/{pkghash}", {"branch": branch}
)
return models.PackageMisconflictBySrcModel(**data)
async def package_name_from_repology(
self, branch: str, name: str
) -> models.PackageNameFromRepologyModel:
data = await self.client.get(
"/site/package_name_from_repology", {"branch": branch, "name": name}
)
return models.PackageNameFromRepologyModel(**data)
async def package_nvr_by_hash(
self, pkghash: int, name: str | None = None
) -> models.PackageNVRByHashModel:
params = {
k: v
for k, v in {
"name": name,
}.items()
if v is not None
}
data = await self.client.get(
f"/site/package_nvr_by_hash/{pkghash}", params or None
)
return models.PackageNVRByHashModel(**data)
async def package_versions(
self,
name: str,
package_type: Literal["source", "binary"] = "source",
arch: str | None = None,
) -> models.SiteSourcePackagesVersionsModel:
params = {
k: v
for k, v in {
"name": name,
"package_type": package_type,
"arch": arch,
}.items()
if v is not None
}
data = await self.client.get("/site/package_versions", params)
return models.SiteSourcePackagesVersionsModel(**data)
async def package_versions_from_images(
self, name: str, branch: str, edition: str, type: str
) -> models.SiteImagePackageVersionsModel:
data = await self.client.get(
"/site/package_versions_from_images",
{"name": name, "branch": branch, "edition": edition, "type": type},
)
return models.SiteImagePackageVersionsModel(**data)
async def package_versions_from_tasks(
self, name: str, branch: str | None = None
) -> models.SItePackagesVersionsFromTasksModel:
params = {
k: v
for k, v in {
"name": name,
"branch": branch,
}.items()
if v is not None
}
data = await self.client.get("/site/package_versions_from_tasks", params)
return models.SItePackagesVersionsFromTasksModel(**data)
async def packagesets_by_hash(
self, pkghash: int
) -> models.SitePackagesetsByHashModel:
data = await self.client.get(f"/site/packagesets_by_hash/{pkghash}")
return models.SitePackagesetsByHashModel(**data)
async def pkghash_by_nvr(
self, name: str, branch: str, version: str, release: str
) -> models.SitePackagesetPackageHashByNameVersionRelease:
data = await self.client.get(
"/site/pkghash_by_nvr",
{"name": name, "branch": branch, "version": version, "release": release},
)
return models.SitePackagesetPackageHashByNameVersionRelease(**data)
async def pkgset_categories_count(
self,
branch: str,
package_type: Literal["all", "source", "binary"] = "source",
) -> models.SitePackagesetCategoriesModel:
data = await self.client.get(
"/site/pkgset_categories_count",
{"branch": branch, "package_type": package_type},
)
return models.SitePackagesetCategoriesModel(**data)
async def pkgsets_summary_status(self) -> models.SitePackagesetsSummaryStatusModel:
data = await self.client.get("/site/pkgsets_summary_status")
return models.SitePackagesetsSummaryStatusModel(**data)
async def repocop_by_maintainer(
self,
branch: str,
maintainer_nickname: str,
by_acl: str = "none",
) -> models.RepocopByMaintainerModel:
data = await self.client.get(
"/site/repocop_by_maintainer",
{
"branch": branch,
"maintainer_nickname": maintainer_nickname,
"by_acl": by_acl,
},
)
return models.RepocopByMaintainerModel(**data)
async def repository_packages(
self,
branch: str,
package_type: Literal["all", "source", "binary"] = "source",
group: str | None = None,
buildtime: int = 0,
) -> models.SitePackagesModel:
params = {
k: v
for k, v in {
"branch": branch,
"package_type": package_type,
"group": group,
"buildtime": buildtime,
}.items()
if v is not None
}
data = await self.client.get("/site/repository_packages", params)
return models.SitePackagesModel(**data)
async def source_package_versions(
self, name: str
) -> models.SiteSourcePackagesVersionsModel:
data = await self.client.get("/site/source_package_versions", {"name": name})
return models.SiteSourcePackagesVersionsModel(**data)
async def tasks_by_maintainer(
self, branch: str, maintainer_nickname: str
) -> models.SiteTaskByNameModel:
data = await self.client.get(
"/site/tasks_by_maintainer",
{"branch": branch, "maintainer_nickname": maintainer_nickname},
)
return models.SiteTaskByNameModel(**data)
async def tasks_by_package(self, name: str) -> models.SiteTaskByNameModel:
data = await self.client.get("/site/tasks_by_package", {"name": name})
return models.SiteTaskByNameModel(**data)
async def tasks_history(
self, task_id: int | None = None
) -> models.SiteTasksHistoryModel:
params = {
k: v
for k, v in {
"task_id": task_id,
}.items()
if v is not None
}
data = await self.client.get("/site/tasks_history", params or None)
return models.SiteTasksHistoryModel(**data)
class AuthInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def login(
self,
nickname: str,
password: str,
auth_provider: Literal["ldap", "keycloak"] = "ldap",
) -> models.AuthResponseModel:
data = await self.client.post(
"/auth/login",
data={
"nickname": nickname,
"password": password,
"auth_provider": auth_provider,
},
)
return models.AuthResponseModel(**data)
async def logout(self) -> models.AuthLogoutResponseModel:
data = await self.client.post("/auth/logout")
return models.AuthLogoutResponseModel(**data)
async def refresh_token(self, access_token: str) -> models.AuthResponseModel:
data = await self.client.post(
"/auth/refresh-token", data={"access_token": access_token}
)
return models.AuthResponseModel(**data)
class DependenciesInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def backport_helper(
self,
from_branch: str,
into_branch: str,
packages_names: list[str],
dp_type: Literal["both", "source", "binary"] = "both",
archs: list[str] | None = None,
) -> models.BackportHelperModel:
params = {
k: v
for k, v in {
"from_branch": from_branch,
"into_branch": into_branch,
"packages_names": packages_names,
"dp_type": dp_type,
"archs": archs,
}.items()
if v is not None
}
data = await self.client.get("/dependencies/backport_helper", params)
return models.BackportHelperModel(**data)
async def binary_package_dependencies(
self, pkghash: int
) -> models.DependenciesPackageDependenciesModel:
data = await self.client.get(
f"/dependencies/binary_package_dependencies/{pkghash}"
)
return models.DependenciesPackageDependenciesModel(**data)
async def fast_lookup(
self, branch: str, dp_name: str, limit: int = 10
) -> models.FastDependencySearchModel:
data = await self.client.get(
"/dependencies/fast_lookup",
{"branch": branch, "dp_name": dp_name, "limit": limit},
)
return models.FastDependencySearchModel(**data)
async def packages_by_dependency(
self,
branch: str,
dp_name: str,
dp_type: Literal["all", "provide", "require", "conflict", "obsolete"] = "all",
last_state: bool = False,
) -> models.DependenciesPackagesModel:
data = await self.client.get(
"/dependencies/packages_by_dependency",
{
"branch": branch,
"dp_name": dp_name,
"dp_type": dp_type,
"last_state": last_state,
},
)
return models.DependenciesPackagesModel(**data)
async def source_package_dependencies(
self, pkghash: int, branch: str, depth: int = 1
) -> models.DependenciesPackageBuildDependenciesModel:
data = await self.client.get(
f"/dependencies/source_package_dependencies/{pkghash}",
{"branch": branch, "depth": depth},
)
return models.DependenciesPackageBuildDependenciesModel(**data)
async def what_depends_src(
self,
name: str,
branch: str,
dp_type: Literal["both", "source", "binary"] = "both",
) -> models.PackageBuildDependencyModel:
data = await self.client.get(
"/dependencies/what_depends_src",
{"name": name, "branch": branch, "dp_type": dp_type},
)
return models.PackageBuildDependencyModel(**data)
class ErrataInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def advisory(
self,
branch: str | None = None,
input: str | None = None,
page: int | None = None,
limit: int | None = None,
sort: list[str] | None = None,
) -> models.AdvisoryErrataModel:
params = {
k: v
for k, v in {
"branch": branch,
"input": input,
"page": page,
"limit": limit,
"sort": sort,
}.items()
if v is not None
}
data = await self.client.get("/errata/advisory", params or None)
return models.AdvisoryErrataModel(**data)
async def branches_updates(
self, errata_ids: list[str], exclude_json: bool = False
) -> models.ErrataBranchesUpdatesModel:
data = await self.client.post(
"/errata/branches_updates", json={"errata_ids": errata_ids}
)
return models.ErrataBranchesUpdatesModel(**data)
async def errata_branches(self) -> models.ErrataBranchesModel:
data = await self.client.get("/errata/errata_branches")
return models.ErrataBranchesModel(**data)
async def export_oval_branches(self) -> models.OvalBranchesModel:
data = await self.client.get("/errata/export/oval/branches")
return models.OvalBranchesModel(**data)
async def export_oval(
self,
branch: str,
package_name: str | None = None,
one_file: bool = False,
) -> bytes:
params = {
k: v
for k, v in {
"package_name": package_name,
"one_file": one_file,
}.items()
if v is not None
}
return await self.client.get_bytes(
f"/errata/export/oval/{branch}", params or None
)
async def find_erratas(
self,
input: list[str] | None = None,
branch: str | None = None,
type: Literal["packages", "repository", "bug", "vuln", "exclusion"]
| None = None,
page: int | None = None,
limit: int | None = None,
state: Literal["all", "active", "discarded"] = "all",
) -> models.ErrataLastChangedModel:
params = {
k: v
for k, v in {
"input": input,
"branch": branch,
"type": type,
"page": page,
"limit": limit,
"state": state,
}.items()
if v is not None
}
data = await self.client.get("/errata/find_erratas", params or None)
return models.ErrataLastChangedModel(**data)
async def find_image_erratas(
self,
uuid: str,
branch: str,
component: str | None = None,
input: list[str] | None = None,
type: Literal["packages", "repository", "bug", "vuln", "exclusion"]
| None = None,
page: int | None = None,
limit: int | None = None,
is_discarded: bool = False,
sort: list[str] | None = None,
) -> models.ImageErrataModel:
params = {
k: v
for k, v in {
"uuid": uuid,
"branch": branch,
"component": component,
"input": input,
"type": type,
"page": page,
"limit": limit,
"is_discarded": is_discarded,
"sort": sort,
}.items()
if v is not None
}
data = await self.client.get("/errata/find_image_erratas", params)
return models.ImageErrataModel(**data)
async def ids(self) -> models.ErrataIdsListModel:
data = await self.client.get("/errata/ids")
return models.ErrataIdsListModel(**data)
async def packages_updates(
self, errata_ids: list[str], exclude_json: bool = False
) -> models.ErrataPackagesUpdatesModel:
data = await self.client.post(
"/errata/packages_updates", json={"errata_ids": errata_ids}
)
return models.ErrataPackagesUpdatesModel(**data)
async def search(
self,
branch: str | None = None,
name: str | None = None,
vuln_id: str | None = None,
errata_id: str | None = None,
) -> models.ErratasModel:
params = {
k: v
for k, v in {
"branch": branch,
"name": name,
"vuln_id": vuln_id,
"errata_id": errata_id,
}.items()
if v is not None
}
data = await self.client.get("/errata/search", params or None)
return models.ErratasModel(**data)
class ExportInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def beehive_ftbfs(
self, branch: str, arch: str | None = None
) -> models.ExportBeehiveFTBFSListModel:
params = {
k: v
for k, v in {
"branch": branch,
"arch": arch,
}.items()
if v is not None
}
data = await self.client.get("/export/beehive/ftbfs", params)
return models.ExportBeehiveFTBFSListModel(**data)
async def branch_binary_packages(
self, branch: str, arch: str | None = None
) -> models.PackagesetPackagesExportModel:
params = {
k: v
for k, v in {
"arch": arch,
}.items()
if v is not None
}
data = await self.client.get(
f"/export/branch_binary_packages/{branch}", params or None
)
return models.PackagesetPackagesExportModel(**data)
async def branch_tree(self) -> models.BranchTreeModel:
data = await self.client.get("/export/branch_tree")
return models.BranchTreeModel(**data)
async def repology(self, branch: str) -> models.RepologyExportModel:
data = await self.client.get(f"/export/repology/{branch}")
return models.RepologyExportModel(**data)
async def sitemap_packages(self, branch: str) -> models.SitemapPackagesExportModel:
data = await self.client.get(f"/export/sitemap_packages/{branch}")
return models.SitemapPackagesExportModel(**data)
async def translation_packages_po_files(
self, branches: list[str], from_date: str | None = None
) -> bytes:
params = {
k: v
for k, v in {
"branches": branches,
"from_date": from_date,
}.items()
if v is not None
}
return await self.client.get_bytes(
"/export/translation/packages_po_files", params
)
class ImageInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def active_images(
self,
branch: str | None = None,
edition: str | None = None,
version: str | None = None,
release: str | None = None,
variant: str | None = None,
type: str | None = None,
) -> models.ActiveImagesModel:
params = {
k: v
for k, v in {
"branch": branch,
"edition": edition,
"version": version,
"release": release,
"variant": variant,
"type": type,
}.items()
if v is not None
}
data = await self.client.get("/image/active_images", params or None)
return models.ActiveImagesModel(**data)
async def find_images_by_package_name(
self,
pkg_name: str,
branch: str | None = None,
edition: str | None = None,
pkg_type: Literal["source", "binary"] = "source",
img_show: Literal["active", "all"] = "all",
) -> models.FindImagesByPackageModel:
params = {
k: v
for k, v in {
"pkg_name": pkg_name,
"branch": branch,
"edition": edition,
"pkg_type": pkg_type,
"img_show": img_show,
}.items()
if v is not None
}
data = await self.client.get("/image/find_images_by_package_name", params)
return models.FindImagesByPackageModel(**data)
async def image_categories_count(
self, uuid: str, component: str | None = None
) -> models.SiteImageCategoriesModel:
params = {
k: v
for k, v in {
"uuid": uuid,
"component": component,
}.items()
if v is not None
}
data = await self.client.get("/image/image_categories_count", params)
return models.SiteImageCategoriesModel(**data)
async def image_info(
self,
branch: str | None = None,
edition: str | None = None,
version: str | None = None,
release: str | None = None,
variant: str | None = None,
flavor: str | None = None,
arch: str | None = None,
component: str | None = None,
platform: str | None = None,
type: str | None = None,
) -> models.ImageInfoModel:
params = {
k: v
for k, v in {
"branch": branch,
"edition": edition,
"version": version,
"release": release,
"variant": variant,
"flavor": flavor,
"arch": arch,
"component": component,
"platform": platform,
"type": type,
}.items()
if v is not None
}
data = await self.client.get("/image/image_info", params or None)
return models.ImageInfoModel(**data)
async def image_packages(
self, uuid: str, group: str | None = None, component: str | None = None
) -> models.ImagePackagesModel:
params = {
k: v
for k, v in {
"uuid": uuid,
"group": group,
"component": component,
}.items()
if v is not None
}
data = await self.client.get("/image/image_packages", params)
return models.ImagePackagesModel(**data)
async def image_packageset(self) -> models.ImagePackageSetModel:
data = await self.client.get("/image/image_packageset")
return models.ImagePackageSetModel(**data)
async def image_status_get(self) -> models.ImageStatusGetModel:
data = await self.client.get("/image/image_status")
return models.ImageStatusGetModel(**data)
async def image_status_post(self, payload: dict) -> None:
await self.client.post("/image/image_status", json=payload)
async def image_tag_status_get(
self, branch: str | None = None, edition: str | None = None
) -> models.ImageTagStatusGetModel:
params = {
k: v
for k, v in {
"branch": branch,
"edition": edition,
}.items()
if v is not None
}
data = await self.client.get("/image/image_tag_status", params or None)
return models.ImageTagStatusGetModel(**data)
async def image_tag_status_post(self, payload: dict) -> None:
await self.client.post("/image/image_tag_status", json=payload)
async def image_uuid_by_tag(self, tag: str) -> models.ImageTagUUIDModel:
data = await self.client.get("/image/image_uuid_by_tag", {"tag": tag})
return models.ImageTagUUIDModel(**data)
async def inspect_regular(
self, payload: dict
) -> models.ImagePackagesInspectRegularModel:
data = await self.client.post("/image/inspect/regular", json=payload)
return models.ImagePackagesInspectRegularModel(**data)
async def inspect_sp(self, payload: dict) -> models.ImagePackagesInspectSPModel:
data = await self.client.post("/image/inspect/sp", json=payload)
return models.ImagePackagesInspectSPModel(**data)
async def iso_all_images(self) -> models.ImageAllISOModel:
data = await self.client.get("/image/iso/all_images")
return models.ImageAllISOModel(**data)
async def last_packages_by_image(
self,
branch: str,
uuid: str,
packages_limit: int | None = None,
component: str | None = None,
) -> models.LastImagePackagesModel:
params = {
k: v
for k, v in {
"branch": branch,
"uuid": uuid,
"packages_limit": packages_limit,
"component": component,
}.items()
if v is not None
}
data = await self.client.get("/image/last_packages_by_image", params)
return models.LastImagePackagesModel(**data)
async def last_packages_image_with_cve_fixed(
self,
branch: str,
uuid: str,
packages_limit: int | None = None,
component: str | None = None,
) -> models.LastImagePackagesModel:
params = {
k: v
for k, v in {
"branch": branch,
"uuid": uuid,
"packages_limit": packages_limit,
"component": component,
}.items()
if v is not None
}
data = await self.client.get(
"/image/last_packages_image_with_cve_fixed", params
)
return models.LastImagePackagesModel(**data)
class LicenseInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def license_text(self) -> str:
return await self.client.get_text("/license")
async def info(self, license: str) -> models.LicenseInfoModel:
data = await self.client.get("/license/info", {"license": license})
return models.LicenseInfoModel(**data)
async def tokens(self, license: str) -> models.LicenseTokensModel:
data = await self.client.get("/license/tokens", {"license": license})
return models.LicenseTokensModel(**data)
class VulnInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def bdu(
self, vuln_id: str, exclude_json: bool = False
) -> models.VulnerabilityInfoModel:
data = await self.client.get(
"/vuln/bdu", {"vuln_id": vuln_id, "exclude_json": exclude_json}
)
return models.VulnerabilityInfoModel(**data)
async def bdu_fixes(
self, vuln_id: str, exclude_json: bool = False
) -> models.VulnFixesPackagesModel:
data = await self.client.get(
"/vuln/bdu/fixes", {"vuln_id": vuln_id, "exclude_json": exclude_json}
)
return models.VulnFixesPackagesModel(**data)
async def cve(
self, vuln_id: str, exclude_json: bool = False
) -> models.VulnerabilityInfoModel:
data = await self.client.get(
"/vuln/cve", {"vuln_id": vuln_id, "exclude_json": exclude_json}
)
return models.VulnerabilityInfoModel(**data)
async def cve_excluded(
self, vuln_id: str, exclude_json: bool = False
) -> models.VulnFixesPackagesModel:
data = await self.client.get(
"/vuln/cve/excluded", {"vuln_id": vuln_id, "exclude_json": exclude_json}
)
return models.VulnFixesPackagesModel(**data)
async def cve_fixes(
self, vuln_id: str, exclude_json: bool = False
) -> models.VulnFixesPackagesModel:
data = await self.client.get(
"/vuln/cve/fixes", {"vuln_id": vuln_id, "exclude_json": exclude_json}
)
return models.VulnFixesPackagesModel(**data)
async def ghsa(
self, vuln_id: str, exclude_json: bool = False
) -> models.VulnerabilityInfoModel:
data = await self.client.get(
"/vuln/ghsa", {"vuln_id": vuln_id, "exclude_json": exclude_json}
)
return models.VulnerabilityInfoModel(**data)
async def ghsa_fixes(
self, vuln_id: str, exclude_json: bool = False
) -> models.VulnFixesPackagesModel:
data = await self.client.get(
"/vuln/ghsa/fixes", {"vuln_id": vuln_id, "exclude_json": exclude_json}
)
return models.VulnFixesPackagesModel(**data)
async def task(self, task_id: int) -> models.CveVulnerableTaskModel:
data = await self.client.get(f"/vuln/task/{task_id}")
return models.CveVulnerableTaskModel(**data)
class ALTRepoAPI:
def __init__(self, session: aiohttp.ClientSession, config: "ALTRepoConfig"):
self.BASE_API_VERSION = "1.25.6"
self._client = BaseAPI(session, config.api_base_url)
self.api = APIInfo(self._client)
self.task = TaskInfo(self._client)
self.package = PackageInfo(self._client)
self.packageset = PackagesetInfo(self._client)
self.acl = AclInfo(self._client)
self.bug = BugInfo(self._client)
self.file = FileInfo(self._client)
self.site = SiteInfo(self._client)
self.auth = AuthInfo(self._client)
self.dependencies = DependenciesInfo(self._client)
self.errata = ErrataInfo(self._client)
self.export = ExportInfo(self._client)
self.image = ImageInfo(self._client)
self.license = LicenseInfo(self._client)
self.vuln = VulnInfo(self._client)
async def version(self) -> models.APIVersion:
return await self.api.version()
from pydantic import BaseModel, Field
from typing import List, Dict, Any
class APIVersion(BaseModel):
name: str
version: str
description: str
class SubTaskArchitectureModel(BaseModel):
stage_status: str
arch: str
class SubTasksElementModel(BaseModel):
subtask_id: int
subtask_type: str
subtask_srpm: str
subtask_srpm_name: str
subtask_srpm_evr: str
subtask_dir: str
subtask_tag_id: str
subtask_tag_name: str
subtask_tag_author: str
subtask_package: str
subtask_pkg_from: str
subtask_changed: str
type: str
archs: List[SubTaskArchitectureModel]
class TaskApprovalElementModel(BaseModel):
type: str
nickname: str
message: str
class TasksListElementModel(BaseModel):
task_id: int
task_repo: str
task_state: str
task_owner: str
task_try: int
task_iter: int
task_testonly: int
task_changed: str
task_message: str
task_stage: str
dependencies: List[int]
subtasks: List[SubTasksElementModel]
approval: list[TaskApprovalElementModel]
class SubTaskInfoElementModel(BaseModel):
subtask_id: int
subtask_type: str
subtask_srpm: str
subtask_srpm_name: str
subtask_srpm_evr: str
subtask_dir: str
subtask_tag_id: str
subtask_tag_name: str
subtask_tag_author: str
subtask_package: str
subtask_pkg_from: str
subtask_changed: str
type: str
src_pkg_name: str
src_pkg_hash: str
archs: List[SubTaskArchitectureModel]
approval: list[TaskApprovalElementModel]
class TaskIterationsElementModel(BaseModel):
task_try: int
task_iter: int
class TaskProgressTaskInfoModel(BaseModel):
task_id: int
task_repo: str
task_state: str
task_owner: str
task_try: int
task_iter: int
task_testonly: int
task_changed: str
task_message: str
task_stage: str
dependencies: List[int]
subtasks: List[SubTaskInfoElementModel]
iterations: list[TaskIterationsElementModel]
class TasksListModel(BaseModel):
request_args: Dict[str, Any]
length: int
tasks: list[TasksListElementModel]
class PackageSetActivePackageSetsModel(BaseModel):
length: int
packagesets: List[str]
class RepositoryStatisticsPackageCountsModel(BaseModel):
arch: str
component: str
count: int
size: int
size_hr: str
uuid: str
class RepositoryStatisticsBranchesModel(BaseModel):
branch: str
date_update: str
packages_count: List[RepositoryStatisticsPackageCountsModel]
class RepositoryStatisticsModel(BaseModel):
length: int
branches: List[RepositoryStatisticsBranchesModel]
class AllMaintainersElementModel(BaseModel):
packager_name: str
packager_nickname: str
count_source_pkg: int
class MaintainerInfoModel(BaseModel):
request_args: Dict[str, Any]
information: AllMaintainersElementModel
class AllMaintainersModel(BaseModel):
request_args: Dict[str, Any]
length: int
maintainers: List[AllMaintainersElementModel]
class SiteWatchByMaintainerElementModel(BaseModel):
pkg_name: str
old_version: str
new_version: str
repology_name: str
url: str
date_update: str
class SiteWatchByMaintainerModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[SiteWatchByMaintainerElementModel]
class BugzillaInfoElementModel(BaseModel):
id: str
status: str
resolution: str
severity: str
product: str
version: str | None
platform: str | None
component: str
source_package_name: str
binary_package_name: str
reporter: str
summary: str
last_changed: str
class BugzillaInfoModel(BaseModel):
request_args: Dict[str, Any]
length: int
bugs: List[BugzillaInfoElementModel]
class FilePackagesByFileElementModel(BaseModel):
hash: str
name: str
version: str
release: str
arch: str
class FilePackagesByFileModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[FilePackagesByFileElementModel]
class PackageInfoChangelogElementModel(BaseModel):
date: str | None
name: str | None
evr: str | None
message: str | None
class PackageInfoDependenciesModel(BaseModel):
require: List[str] | None
provide: List[str] | None
conflict: List[str] | None
obsolete: List[str] | None
class PackageInfoPackageModel(BaseModel):
name: str
version: str
release: str
sha1: str
packager: str
packager_email: str
arch: str
epoch: int
disttag: str
sourcepackage: int
filename: str
sourcerpm: str
serial: int | None
buildtime: int | None
buildhost: str | None
size: int | None
archivesize: int | None
filesize: int | None
rpmversion: str | None
cookie: str | None
license: str | None
group: str | None
url: str | None
summary: str | None
description: str | None
distribution: str | None
vendor: str | None
os: str | None
gif: str | None
xpm: str | None
icon: str | None
prein: str | None
postin: str | None
preun: str | None
postun: str | None
preinprog: List[str] | None
postinprog: List[str] | None
preunprog: List[str] | None
postunprog: List[str] | None
buildarchs: List[str] | None
verifyscript: str | None
verifyscriptprog: List[str] | None
prefixes: List[str] | None
instprefixes: List[str] | None
optflags: str | None
disturl: str | None
payloadformat: str | None
payloadcompressor: str | None
payloadflags: str | None
platform: str | None
changelog: List[PackageInfoChangelogElementModel] | PackageInfoChangelogElementModel
files: List[str] | None
depends: List[PackageInfoDependenciesModel] | PackageInfoDependenciesModel
class PackageInfoModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[PackageInfoPackageModel]
class PackageByFileNameElementModel(BaseModel):
name: str
version: str
release: str
disttag: str
sha1: str
branch: str
arch: str
files: List[str]
class PackagesByFileNamesJsonModel(BaseModel):
files: List[str]
branch: str
arch: str | None = None
class PackageByFileNameModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[PackageByFileNameElementModel]
not_found: List[str]
class FilesElementModel(BaseModel):
file_name: str
file_hashname: str
file_class: str
symlink: str
file_mode: str
class FilesModel(BaseModel):
request_args: Dict[str, Any]
length: int
files: List[FilesElementModel]
class AclGroupsElementModel(BaseModel):
group: str
date: str
maintainers: List[str]
class AclGroupsModel(BaseModel):
request_args: Dict[str, Any]
length: int
groups: List[AclGroupsElementModel]
class SitePackageInfoArchsModel(BaseModel):
name: str
archs: list[str]
pkghash: list[str]
class SitePackageTasksElementModel(BaseModel):
type: str
id: int
date: str
class SitePackageInfoChangelogElementModel(BaseModel):
date: str
name: str
nick: str
evr: str
message: str
class SitePackageNewVersionModel(BaseModel):
task_id: int
date: str
pkghash: str
version: str
release: str
class SitePackageBeehiveElementModel(BaseModel):
arch: str
status: str
updated: str
build_time: float
ftbfs_since: str
url: str
class SitePackageDependenciesElementModel(BaseModel):
name: str
version: str
type: str
flag: int
flag_decoded: List[str]
class SitePackageLicenseTokensElementModel(BaseModel):
token: str
license: str
class SitePackageInfoModel(BaseModel):
pkghash: str
request_args: Dict[str, Any]
name: str
version: str
release: str
arch: str
buildtime: int
task: int
task_date: str
gear: str
license: str
category: str
url: str
vcs: str
summary: str
description: str
packager: str
packager_nickname: str
acl: List[str]
maintainers: List[str]
package_archs: List[SitePackageInfoArchsModel]
tasks: List[SitePackageTasksElementModel]
changelog: List[SitePackageInfoChangelogElementModel]
new_version: List[SitePackageNewVersionModel]
beehive: List[SitePackageBeehiveElementModel]
dependencies: List[SitePackageDependenciesElementModel]
license_tokens: List[SitePackageLicenseTokensElementModel]
class SitePackagesetPackageHashModel(BaseModel):
request_args: Dict[str, Any]
pkghash: str
version: str
release: str
class FindSourcePackageInBranch(BaseModel):
request_args: Dict[str, Any]
source_package: str
class AclByPackagesElementModel(BaseModel):
name: str
updated: str
members: List[str]
class AclByPackagesModel(BaseModel):
branch: str
packages: List[AclByPackagesElementModel]
class AclMaintainerGroupsElementModel(BaseModel):
name: str
groups: List[str]
class AclMaintainerGroupsModel(BaseModel):
nickname: str
branches: List[AclMaintainerGroupsElementModel]
class AuthResponseModel(BaseModel):
access_token: str
refresh_token: str
class AuthLogoutResponseModel(BaseModel):
message: str
class BackportHelperBinaryElementModel(BaseModel):
srpm: str
name: str
epoch: int
version: str
release: str
arch: str
class BackportHelperBinaryDepthElementModel(BaseModel):
depth: int
packages: List[BackportHelperBinaryElementModel]
class BackportHelperModel(BaseModel):
request_args: Dict[str, Any]
count: int
maxdepth: int
dependencies: List[BackportHelperBinaryDepthElementModel]
class DependenciesPackageDependenciesElementModel(BaseModel):
name: str
version: str
type: str
flag: int
flag_decoded: List[str]
class DependenciesPackageDependenciesModel(BaseModel):
request_args: Dict[str, Any]
length: int
dependencies: List[DependenciesPackageDependenciesElementModel]
class DependenciesPackageInfoModel(BaseModel):
name: str
epoch: int
version: str
release: str
buildtime: int
class DependenciesPackageInfoElementModel(BaseModel):
name: str
version: str
release: str
summary: str
pkghash: str
class DependenciesPackageBuildDependenciesModel(BaseModel):
request_args: Dict[str, Any]
length: int
package_info: DependenciesPackageInfoModel | None
dependencies: List[DependenciesPackageDependenciesElementModel]
provided_by_src: List[DependenciesPackageInfoElementModel]
class FastDependencySearchElementModel(BaseModel):
dp_name: str
class FastDependencySearchModel(BaseModel):
request_args: Dict[str, Any]
length: int
dependencies: List[FastDependencySearchElementModel]
class DependenciesAllPackagasetsElementModel(BaseModel):
branch: str
count: int
class DependenciesPackagesElementModel(BaseModel):
hash: str
name: str
version: str
release: str
arch: str
sourcepackage: int
summary: str
buildtime: int
category: str
maintainer: str
dp_types: List[str]
class DependenciesPackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[DependenciesPackagesElementModel]
branches: List[DependenciesAllPackagasetsElementModel]
class PackageDependsElementModel(BaseModel):
pkghash: str
name: str
arch: str
dp_name: str
dp_version: str
dp_flag: int
dp_flag_decoded: List[str]
class PackageDependsModel(BaseModel):
requires: PackageDependsElementModel | None
provides: PackageDependsElementModel | None
class PackageBuildDependencyElementModel(BaseModel):
pkghash: str
name: str
branch: str
buildtime: str
acl: List[str]
depends: List[PackageDependsModel]
class PackageBuildDependencyModel(BaseModel):
request_args: Dict[str, Any]
length: int
dependencies: List[PackageBuildDependencyElementModel]
class AdvisoryErrataReferenceModel(BaseModel):
type: str
link: str
class AdvisoryErrataPackageVersionRangeModel(BaseModel):
begin: str
begin_exclude: bool
end: str
end_exclude: bool
class AdvisoryErrataJsonModel(BaseModel):
type: str
action: str
is_public: bool
reason: str
description: str
vuln_id: str
vuln_cpe: str
branches: List[str]
pkg_name: str
pkg_evr: str
pkg_versions: List[AdvisoryErrataPackageVersionRangeModel]
references: List[AdvisoryErrataReferenceModel]
extra: Dict[str, Any]
class AdvisoryErrataElementModel(BaseModel):
model_config = {"populate_by_name": True}
id: str
type: str
created: str
updated: str
pkgset_name: str
task_id: int
subtask_id: int
task_state: str
pkg_hash: str
pkg_name: str
pkg_version: str
pkg_release: str
references: List[AdvisoryErrataReferenceModel]
json_: AdvisoryErrataJsonModel | None = Field(None, alias="json")
class AdvisoryErrataModel(BaseModel):
request_args: Dict[str, Any]
length: int
erratas: List[AdvisoryErrataElementModel]
class ErrataBranchesModel(BaseModel):
length: int
branches: List[str]
class ErrataJsonPostListModel(BaseModel):
errata_ids: List[str]
class ErrataBugModel(BaseModel):
id: int
summary: str
is_valid: bool
class ErrataVulnerabilityModel(BaseModel):
id: str
hash: str
type: str
summary: str
score: float
severity: str
url: str
modified_date: str
published_date: str
body: str
is_valid: bool
parsed: Dict[str, Any] | None
class ErrataPackageUpdateModel(BaseModel):
id: str
type: str
created: str
updated: str
pkgset_name: str
task_id: int
subtask_id: int
task_state: str
pkg_hash: str
pkg_name: str
pkg_version: str
pkg_release: str
bugs: List[ErrataBugModel]
vulns: List[ErrataVulnerabilityModel]
class ErrataBranchUpdateModel(BaseModel):
id: str
type: str
pkgset_name: str
pkgset_date: str
packages_updates: List[ErrataPackageUpdateModel]
class ErrataBranchesUpdatesModel(BaseModel):
branches_updates: List[ErrataBranchUpdateModel]
class ErrataPackagesUpdatesModel(BaseModel):
packages_updates: List[ErrataPackageUpdateModel]
class OvalBranchesModel(BaseModel):
length: int
branches: List[str]
class VulnerabilitiesElementModel(BaseModel):
id: str
type: str
class PackagesElementModel(BaseModel):
pkghash: str
pkg_name: str
pkg_version: str
pkg_release: str
class ErrataLastChangedElementModel(BaseModel):
model_config = {"populate_by_name": True}
is_discarded: bool
errata_id: str
eh_type: str
task_id: int
changed: str
branch: str
json_: Dict[str, Any] = Field(default_factory=dict, alias="json")
packages: List[PackagesElementModel]
vulnerabilities: List[VulnerabilitiesElementModel]
class ErrataLastChangedModel(BaseModel):
request_args: Dict[str, Any]
length: int
erratas: List[ErrataLastChangedElementModel]
class ImageErrataElementModel(BaseModel):
img_hash: str
img_version: str
img_release: str
pkg_name: str
pkg_arch: str
pkg_hash: str
pkg_version: str
pkg_release: str
summary: str
errata_id: str
eh_type: str
task_id: int
changed: str
branch: str
is_discarded: bool
vulnerabilities: List[VulnerabilitiesElementModel]
class ImageErrataModel(BaseModel):
request_args: Dict[str, Any]
length: int
erratas: List[ImageErrataElementModel]
class ErrataIdsListModel(BaseModel):
errata_ids: List[str]
class ErrataReferenceModel(BaseModel):
id: str
type: str
class ErrataModel(BaseModel):
id: str
type: str
created: str
updated: str
pkgset_name: str
task_id: int
subtask_id: int
task_state: str
pkg_hash: str
pkg_name: str
pkg_version: str
pkg_release: str
references: List[ErrataReferenceModel]
class ErratasModel(BaseModel):
erratas: List[ErrataModel]
class ExportBeehiveFTBFSElementModel(BaseModel):
branch: str
hash: str
name: str
epoch: int
version: str
release: str
arch: str
updated: str
ftbfs_since: str
url: str
class ExportBeehiveFTBFSListModel(BaseModel):
request_args: Dict[str, Any]
length: int
ftbfs: List[ExportBeehiveFTBFSElementModel]
class PackagesetPackagesExportElementModel(BaseModel):
name: str
epoch: int
version: str
release: str
arch: str
disttag: str
buildtime: int
source: str
class PackagesetPackagesExportModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[PackagesetPackagesExportElementModel]
class BranchTreeTaskModel(BaseModel):
id: int
prev: int
branch: str
date: str
class BranchTreeBranchCommitModel(BaseModel):
name: str
date: str
task: int
class BranchTreeBranchPointModel(BaseModel):
branch: str
task: BranchTreeTaskModel | None
from_task: BranchTreeTaskModel | None
class BranchTreeModel(BaseModel):
branches: List[str]
tasks: List[BranchTreeTaskModel]
branch_commits: List[BranchTreeBranchCommitModel]
branch_points: List[BranchTreeBranchPointModel]
class RepologyExportBranchBinaryPackageElementModel(BaseModel):
name: str
epoch: int
version: str
release: str
summary: str
archs: List[str]
class RepologyExportBranchSourcePackageElementModel(BaseModel):
name: str
epoch: int
version: str
release: str
url: str
license: str
category: str
summary: str
packager: str
homepage: str
recipe: str
recipe_raw: str
bugzilla: str
CPE: List[str]
binaries: List[RepologyExportBranchBinaryPackageElementModel]
class RepologyExportBranchStatElementModel(BaseModel):
arch: str
count: int
class RepologyExportModel(BaseModel):
branch: str
date: str
stats: List[RepologyExportBranchStatElementModel]
packages: List[RepologyExportBranchSourcePackageElementModel]
class SitemapPackagesElementModel(BaseModel):
pkghash: str
name: str
buildtime: int
class SitemapPackagesExportModel(BaseModel):
branch: str
packages: List[SitemapPackagesElementModel]
class FastFileSearchElementModel(BaseModel):
file_name: str
class FastFileSearchModel(BaseModel):
request_args: Dict[str, Any]
length: int
files: List[FastFileSearchElementModel]
class ActiveImagesElementModel(BaseModel):
edition: str
tags: List[str]
class ActiveImagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
images: List[ActiveImagesElementModel]
class FindImagesByPackageElementModel(BaseModel):
pkghash: str
name: str
branch: str
version: str
release: str
arch: str
edition: str
tag: str
file: str
date: str
class FindImagesByPackageModel(BaseModel):
request_args: Dict[str, Any]
length: int
images: List[FindImagesByPackageElementModel]
class SiteImageCategoryElementModel(BaseModel):
category: str
count: int
class SiteImageCategoriesModel(BaseModel):
request_args: Dict[str, Any]
length: int
categories: List[SiteImageCategoryElementModel]
class ImageInfoComponentModel(BaseModel):
name: str
size: str
packages: int
uuid: str
ruuid: str
kv: Dict[str, Any]
class ImageInfoElementModel(BaseModel):
date: str
uuid: str
tag: str
branch: str
edition: str
flavor: str
platform: str
release: str
version_major: int
version_minor: int
version_sub: int
arch: str
variant: str
type: str
file: str
url: List[str]
md5sum: str
gost12sum: str
sha256sum: str
components: List[ImageInfoComponentModel]
class ImageInfoModel(BaseModel):
request_args: Dict[str, Any]
length: int
images: List[ImageInfoElementModel]
class ImagePackagesElementModel(BaseModel):
hash: str
name: str
version: str
release: str
arch: str
summary: str
buildtime: int
changelog_text: str
class ImagePackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
subcategories: List[str]
packages: List[ImagePackagesElementModel]
class ImagePackageSetModel(BaseModel):
length: int
branches: List[str]
class ImageStatusGetElementModel(BaseModel):
model_config = {"populate_by_name": True}
branch: str
edition: str
name: str
show: str
start_date: str
end_date: str
summary_ru: str
summary_en: str
description_ru: str
description_en: str
mailing_list: str
name_bugzilla: str
json_: Dict[str, Any] = Field(default_factory=dict, alias="json")
class ImageStatusGetModel(BaseModel):
images: List[ImageStatusGetElementModel]
class ImageJSONElementModel(BaseModel):
img_edition: str
img_name: str
img_show: str
img_summary_ru: str
img_summary_en: str
img_start_date: str
img_end_date: str
img_mailing_list: str
img_name_bugzilla: str
img_json: Dict[str, Any]
class ImageJSONModel(BaseModel):
img_branch: str
img_description_ru: str
img_description_en: str
images: List[ImageJSONElementModel]
class ImageTagStatusGetElementModel(BaseModel):
tag: str
show: str
class ImageTagStatusGetModel(BaseModel):
tags: List[ImageTagStatusGetElementModel]
class ImageTagJSONElementModel(BaseModel):
img_tag: str
img_show: str
class ImageTagJSONModel(BaseModel):
tags: List[ImageTagJSONElementModel]
class ImageTagUUIDModel(BaseModel):
request_args: Dict[str, Any]
uuid: str
file: str
type: str
components: Dict[str, Any]
class ImagePackagesElement1Model(BaseModel):
hash: str
arch: str
name: str
epoch: int
version: str
release: str
disttag: str
buildtime: int
class ImagePackagesElement2Model(ImagePackagesElement1Model):
task_id: int
subtask_id: int
class ImagePackagesJSONElementModel(BaseModel):
pkg_hash: str
pkg_name: str
pkg_epoch: int
pkg_version: str
pkg_release: str
pkg_arch: str
pkg_disttag: str
pkg_buildtime: int
class ImagePackagesJSONModel(BaseModel):
branch: str
packages: List[ImagePackagesJSONElementModel]
class ImagePackagesInspectSPPackageModel(BaseModel):
found_in: str
version_check: str
image: ImagePackagesElement1Model | None
database: ImagePackagesElement1Model | None
last_branch: ImagePackagesElement1Model | None
class ImagePackagesInspectRegularModel(BaseModel):
request_args: Dict[str, Any]
input_pakages: int
not_in_branch: int
found_in_tasks: int
not_found_in_db: int
packages_in_tasks: List[ImagePackagesElement2Model]
packages_not_in_db: List[ImagePackagesElement1Model]
class ImagePackagesInspectSPModel(BaseModel):
request_args: Dict[str, Any]
input_pakages: int
in_branch: int
not_in_branch: int
found_in_tasks: int
not_found_in_db: int
packages: List[ImagePackagesInspectSPPackageModel]
class ImageAllISOElementModel(BaseModel):
branch: str
name: str
tag: str
file: str
date: str
uuid: str
class ImageAllISOModel(BaseModel):
length: int
images: List[ImageAllISOElementModel]
class LastImagePackagesElementModel(BaseModel):
task_id: str
task_changed: str
tplan_action: str
branch: str
hash: str
name: str
version: str
release: str
arch: str
img_hash: str
img_version: str
img_release: str
summary: str
chlog_name: str
chlog_nick: str
chlog_date: str
chlog_text: str
class LastImagePackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[LastImagePackagesElementModel]
class FindImagesByTaskImageElementModel(BaseModel):
filename: str
edition: str
tag: str
buildtime: str
binpkg_name: str
binpkg_version: str
binpkg_release: str
binpkg_arch: str
binpkg_hash: str
class FindImagesByTaskIterationElementModel(BaseModel):
task_try: int
task_iter: int
class FindImagesByTaskSubtaskElementModel(BaseModel):
id: int
type: str
srpm_name: str
srpm_hash: str
pkg_version: str
pkg_release: str
images: List[FindImagesByTaskImageElementModel]
class FindImagesByTaskModel(BaseModel):
task_id: int
task_state: str
task_testonly: int
task_repo: str
task_owner: str
task_try: int
task_iter: int
task_message: str
task_changed: str
dependencies: List[int]
subtasks: List[FindImagesByTaskSubtaskElementModel]
iterations: List[FindImagesByTaskIterationElementModel]
class CheckImagesInputFilterModel(BaseModel):
editions: List[str]
releases: List[str]
versions: List[str]
archs: List[str]
variants: List[str]
types: List[str]
class CheckImagesInputModel(BaseModel):
task_id: int
packages_names: List[str]
filters: List[CheckImagesInputFilterModel]
class CheckImagesOutputPackageModel(BaseModel):
status: str
from_subtask: int
srcpkg_name: str
binpkg_name: str
binpkg_arch: str
class CheckImagesOutputImageModel(BaseModel):
file: str
branch: str
edition: str
flavor: str
platform: str
release: str
major_version: int
minor_version: int
sub_version: int
arch: str
variant: str
type: str
buildtime: str
packages: List[CheckImagesOutputPackageModel]
class CheckImagesOutputModel(BaseModel):
request_args: Dict[str, Any]
in_images: List[CheckImagesOutputImageModel]
not_in_images: List[CheckImagesOutputPackageModel]
class LicenseInfoModel(BaseModel):
request_args: Dict[str, Any]
id: str
name: str
text: str
type: str
header: str
comment: str
urls: List[str]
class LicenseTokensElementModel(BaseModel):
token: str
license: str
class LicenseTokensModel(BaseModel):
request_args: Dict[str, Any]
length: int
tokens: List[LicenseTokensElementModel]
class BuildDependencySetAmbiguousProvidesPackageModel(BaseModel):
name: str
used: bool
class BuildDependencySetAmbiguousProvidesElementModel(BaseModel):
requires: str
provides_count: int
provides: List[BuildDependencySetAmbiguousProvidesPackageModel]
class BuildDependencySetAmbiguousProvidesModel(BaseModel):
package: str
ambiguous_provides: List[BuildDependencySetAmbiguousProvidesElementModel]
class BuildDependencySetPackageModel(BaseModel):
name: str
version: str
release: str
epoch: int
archs: List[str]
requires: List[str]
class BuildDependencySetPackagesModel(BaseModel):
package: str
length: int
depends: List[BuildDependencySetPackageModel]
class BuildDependencySetModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[BuildDependencySetPackagesModel]
ambiguous_dependencies: List[BuildDependencySetAmbiguousProvidesModel]
class PackageFindPackagesetElementModel(BaseModel):
branch: str
pkgset_datetime: str
sourcepkgname: str
packages: List[str]
version: str
release: str
disttag: str
packager_email: str
buildtime: str
archs: List[str]
class PackageFindPackagesetModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[PackageFindPackagesetElementModel]
class MaintainerScoreElementModel(BaseModel):
nick: str
score: float
base_score: float
updates: int
patches: int
nmu: int
bugfixes: int
bugfixes_with_update: int
in_acl: bool
last_activity: str
recent_commits: int
bonus_applied: bool
class MaintainerScoreModel(BaseModel):
request_args: Dict[str, Any]
package: str
branch: str
primary_maintainer: str
status: str
maintainers: List[MaintainerScoreElementModel]
class PackageMisconflictPackageModel(BaseModel):
input_package: str
conflict_package: str
version: str
release: str
epoch: int
archs: List[str]
files_with_conflict: List[str]
class PackageMisconflictPackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
conflicts: List[PackageMisconflictPackageModel]
class PackageFilesElementModel(BaseModel):
file_name: str
file_size: str
file_class: str
symlink: str
file_mtime: str
file_mode: str
class PackageFilesModel(BaseModel):
request_args: Dict[str, Any]
length: int
files: List[PackageFilesElementModel]
class RepocopJsonGetModel(BaseModel):
pkg_name: str
pkg_version: str
pkg_release: str
pkg_arch: str
branch: str
test_name: str
test_status: str
test_message: str
test_date: str
class RepocopJsonGetListModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[RepocopJsonGetModel]
class RepocopJsonPostModel(BaseModel):
pkg_name: str
pkg_version: str
pkg_release: str
pkg_arch: str
pkgset_name: str
rc_srcpkg_name: str
rc_srcpkg_version: str
rc_srcpkg_release: str
rc_test_name: str
rc_test_status: str
rc_test_message: str
rc_test_date: str
class RepocopJsonPostListModel(BaseModel):
packages: List[RepocopJsonPostModel]
class PackageSpecfileModel(BaseModel):
request_args: Dict[str, Any]
pkg_hash: str
pkg_name: str
pkg_version: str
pkg_release: str
specfile_name: str
specfile_date: str
specfile_content: str
class UnpackagedDirsElementModel(BaseModel):
package: str
directory: str
version: str
release: str
epoch: int
packager: str
email: str
archs: List[str]
class UnpackagedDirsModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[UnpackagedDirsElementModel]
class PackagesetComparePackageModel(BaseModel):
name: str
version: str
release: str
class PackagesetCompareElementModel(BaseModel):
pkgset1: str
pkgset2: str
package1: PackagesetComparePackageModel | None
package2: PackagesetComparePackageModel | None
class PackagesetCompareModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[PackagesetCompareElementModel]
class MaintainerScoresBatchMaintainerModel(BaseModel):
nick: str
score: float
base_score: float
updates: int
patches: int
nmu: int
bugfixes: int
in_acl: bool
last_activity: str
recent_commits: int
bonus_applied: bool
class MaintainerScoresBatchElementModel(BaseModel):
package: str
primary_maintainer: str
status: str
maintainers: List[MaintainerScoresBatchMaintainerModel]
class MaintainerScoresBatchModel(BaseModel):
request_args: Dict[str, Any]
branch: str
length: int
packages: List[MaintainerScoresBatchElementModel]
class PackagesByUuidElementModel(BaseModel):
hash: str
name: str
version: str
release: str
arch: str
sourcerpm: str
summary: str
buildtime: int
changelog_text: str
class PackagesByUuidModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[PackagesByUuidElementModel]
class PackageSetStatusGetElementModel(BaseModel):
branch: str
pkgset_name_bugzilla: str
start_date: str
end_date: str
show: int
description_ru: str
description_en: str
url_mailing_list: str
mirrors_json: Dict[str, Any]
has_images: int
class PackageSetStatusGetModel(BaseModel):
branches: List[PackageSetStatusGetElementModel]
class PackageSetStatusPostElementModel(BaseModel):
pkgset_name: str
rs_pkgset_name_bugzilla: str
rs_start_date: str
rs_end_date: str
rs_show: int
rs_description_ru: str
rs_description_en: str
rs_mailing_list: str
rs_mirrors_json: List[Dict[str, Any]]
class PackageSetStatusPostModel(BaseModel):
branches: List[PackageSetStatusPostElementModel]
class PackagesetPackagesElementModel(BaseModel):
hash: str
name: str
version: str
release: str
summary: str
url: str
license: str
category: str
maintainers: List[str]
acl_list: List[str]
archs: List[str]
class PackagesetPackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[PackagesetPackagesElementModel]
done_tasks: List[int]
class SiteAllArchsElementModel(BaseModel):
arch: str
count: int
class SiteAllArchsModel(BaseModel):
length: int
archs: List[SiteAllArchsElementModel]
class SiteAllPackagasetsElementModel(BaseModel):
branch: str
count: int
class SiteAllPackagasetsModel(BaseModel):
length: int
branches: List[SiteAllPackagasetsElementModel]
class SiteAllPackagesetsSummaryCountsModel(BaseModel):
arch: str
count: int
class SiteAllPackagesetsSummaryBranchesModel(BaseModel):
branch: str
packages_count: List[SiteAllPackagesetsSummaryCountsModel]
class SiteAllPackagesetsSummaryModel(BaseModel):
length: int
branches: List[SiteAllPackagesetsSummaryBranchesModel]
class SiteBeehiveByMaintainerElementModel(BaseModel):
branch: str
name: str
version: str
release: str
arch: str
updated: str
ftbfs_since: str
build_time: float
url: str
class SiteBeehiveByMaintainerModel(BaseModel):
request_args: Dict[str, Any]
length: int
beehive: List[SiteBeehiveByMaintainerElementModel]
class SitePackageVersionsElementModel(BaseModel):
branch: str
version: str
release: str
pkghash: str
deleted: bool
class SitePackagesBinaryListElementModel(BaseModel):
hash: str
name: str
version: str
release: str
arch: str
class SitePackagesBinaryListModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[SitePackagesBinaryListElementModel]
versions: List[SitePackageVersionsElementModel]
class SiteBinPackageScriptsElementModel(BaseModel):
prein: str
postin: str
preun: str
postun: str
pretrans: str
posttrans: str
class SiteBinPackageScriptsModel(BaseModel):
request_args: Dict[str, Any]
pkg_name: str
pkg_arch: str
length: int
scripts: List[SiteBinPackageScriptsElementModel]
class SiteDeletedPackageModel(BaseModel):
branch: str
package: str
version: str
release: str
hash: str
task_id: int
subtask_id: int
task_owner: str
subtask_owner: str
task_changed: str
task_message: str
class SiteFastPackagesSearchElementModel(BaseModel):
name: str
sourcepackage: str
branches: List[str]
class SiteFastPackagesSearchModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[SiteFastPackagesSearchElementModel]
class SiteFingPackagesPackageModel(BaseModel):
name: str
buildtime: int
url: str
summary: str
category: str
versions: List[SitePackageVersionsElementModel]
by_binary: bool
class SiteFingPackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[SiteFingPackagesPackageModel]
class SiteLastPackagesElementModel(BaseModel):
subtask_id: int
subtask_userid: str
subtask_type: str
hash: str
name: str
version: str
release: str
summary: str
buildtime: int
changelog_name: str
changelog_nickname: str
changelog_date: str
changelog_text: str
class SiteLastPackagesPackageModel(BaseModel):
task_id: int
task_owner: str
task_changed: str
task_message: str
packages: List[SiteLastPackagesElementModel]
class SiteLastPackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
tasks: List[SiteLastPackagesPackageModel]
last_branch_task: int
last_branch_date: str
class SiteLastBranchPackagesPackageModel(BaseModel):
hash: str
name: str
version: str
release: str
summary: str
buildtime: int
changelog_name: str
changelog_nickname: str
changelog_date: str
changelog_text: str
class SiteLastBranchPackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[SiteLastBranchPackagesPackageModel]
last_branch_date: str
class SiteLastPackagesWithCVEFixesElementModel(BaseModel):
hash: str
name: str
version: str
release: str
summary: str
buildtime: int
changelog_date: str
changelog_text: str
class SiteLastPackagesWithCVEFixesModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[SiteLastPackagesWithCVEFixesElementModel]
class MaintainerBranchesModel(BaseModel):
request_args: Dict[str, Any]
length: int
branches: List[SiteAllPackagasetsElementModel]
class MaintainerPackagesElementModel(BaseModel):
name: str
buildtime: int
url: str
summary: str
version: str
release: str
class MaintainerPackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[MaintainerPackagesElementModel]
class SiteChangelogElementModel(BaseModel):
date: str
name: str
nick: str
evr: str
message: str
class SiteChangelogModel(BaseModel):
pkghash: str
request_args: Dict[str, Any]
length: int
changelog: List[SiteChangelogElementModel]
class SitePackagesDownloadsPackageModel(BaseModel):
name: str
url: str
md5: str
size: str
class SitePackagesDownloadsElementModel(BaseModel):
arch: str
packages: List[SitePackagesDownloadsPackageModel]
class SitePackagesDownloadsModel(BaseModel):
pkghash: str
request_args: Dict[str, Any]
downloads: List[SitePackagesDownloadsElementModel]
class SiteBriefPackageInfoModel(BaseModel):
name: str
version: str
release: str
arch: str
summary: str
type: str
class BinPackageLogElementModel(BaseModel):
pkg_hash: str
task_id: int
subtask_id: int
subtask_arch: str
buildlog_hash: str
link: str
class PackageMisconflictBySrcElementModel(BaseModel):
input_package: str
input_archs: List[str]
conflict_package: str
version: str
release: str
epoch: int
archs: List[str]
files_with_conflict: List[str]
explicit: bool
class PackageMisconflictBySrcModel(BaseModel):
request_args: Dict[str, Any]
length: int
conflicts: List[PackageMisconflictBySrcElementModel]
class PackageNameFromRepologyModel(BaseModel):
request_args: Dict[str, Any]
name: str
repo: str
class PackageNVRByHashModel(BaseModel):
request_args: Dict[str, Any]
hash: str
name: str
version: str
release: str
is_source: bool
class SiteSourcePackagesVersionsModel(BaseModel):
request_args: Dict[str, Any]
versions: List[SitePackageVersionsElementModel]
class SiteImagePackageVersionsElementModel(BaseModel):
tag: str
uuid: str
version_major: int
version_minor: int
version_sub: int
img_arch: str
platform: str
variant: str
flavor: str
type: str
hash: str
name: str
version: str
release: str
arch: str
class SiteImagePackageVersionsModel(BaseModel):
request_args: Dict[str, Any]
length: int
versions: List[SiteImagePackageVersionsElementModel]
class SItePackagesVersionsFromTasksElementModel(BaseModel):
branch: str
task: int
hash: str
owner: str
changed: str
name: str
version: str
release: str
class SItePackagesVersionsFromTasksModel(BaseModel):
request_args: Dict[str, Any]
length: int
versions: List[SItePackagesVersionsFromTasksElementModel]
class SitePackagesetsByHashModel(BaseModel):
pkghash: str
length: int
branches: List[str]
class SitePackagesetPackageHashByNameVersionRelease(BaseModel):
request_args: Dict[str, Any]
pkghash: str
class SitePackagesetCategoryElementModel(BaseModel):
category: str
count: int
class SitePackagesetCategoriesModel(BaseModel):
request_args: Dict[str, Any]
length: int
categories: List[SitePackagesetCategoryElementModel]
class SitePackagesetStatusElementModel(BaseModel):
branch: str
start_date: str
end_date: str
show: int
description_ru: str
description_en: str
has_images: int
class SitePackagesetsSummaryStatusModel(BaseModel):
length: int
branches: List[SiteAllPackagesetsSummaryBranchesModel]
status: List[SitePackagesetStatusElementModel]
class RepocopByMaintainerElementModel(BaseModel):
pkg_name: str
pkg_version: str
pkg_release: str
pkg_arch: str
srcpkg_name: str
branch: str
test_name: str
test_status: str
test_message: str
test_date: str
class RepocopByMaintainerModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[RepocopByMaintainerElementModel]
class SitePackagesElementModel(BaseModel):
hash: str
name: str
version: str
release: str
summary: str
buildtime: int
category: str
maintainer: str
changelog: str
task_id: int
subtask_id: int
task_owner: str
class SitePackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
subcategories: List[str]
packages: List[SitePackagesElementModel]
class SiteTaskByNamePackageModel(BaseModel):
type: str
name: str
version: str
release: str
link: str
class SiteTaskByNameTaskModel(BaseModel):
id: int
state: str
branch: str
owner: str
changed: str
packages: List[SiteTaskByNamePackageModel]
class SiteTaskByNameModel(BaseModel):
request_args: Dict[str, Any]
length: int
tasks: List[SiteTaskByNameTaskModel]
class TasksHistoryBranchCommitModel(BaseModel):
name: str
date: str
task: int
class TasksHistoryTaskModel(BaseModel):
id: int
prev: int
branch: str
date: str
class SiteTasksHistoryModel(BaseModel):
branches: List[str]
tasks: List[TasksHistoryTaskModel]
branch_commits: List[TasksHistoryBranchCommitModel]
class AllTasksBranchesModel(BaseModel):
length: int
branches: List[str]
class FindTasksElementModel(BaseModel):
task_id: int
task_owner: str
task_state: str
task_repo: str
components: List[str]
class FindTasksModel(BaseModel):
request_args: Dict[str, Any]
length: int
tasks: List[FindTasksElementModel]
class TaskInfoPackageModel(BaseModel):
name: str
version: str
release: str
filename: str
class TaskInfoApprovalsModel(BaseModel):
date: str
type: str
name: str
message: str
class TaskInfoArchsModel(BaseModel):
last_changed: str
arch: str
status: str
class TaskInfoSubtaskModel(BaseModel):
subtask_id: int
last_changed: str
userid: str
type: str
sid: str
dir: str
package: str
tag_author: str
tag_name: str
tag_id: str
srpm: str
srpm_name: str
srpm_evr: str
pkg_from: str
source_package: TaskInfoPackageModel | None
approvals: List[TaskInfoApprovalsModel]
archs: List[TaskInfoArchsModel]
class TaskInfoPlanModel(BaseModel):
src: List[TaskInfoPackageModel]
bin: List[TaskInfoPackageModel]
class TaskInfoPlan2Model(BaseModel):
model_config = {"populate_by_name": True}
add: TaskInfoPlanModel | None
del_: TaskInfoPlanModel | None = Field(None, alias="del")
class TaskInfoModel(BaseModel):
model_config = {"populate_by_name": True}
id: int
prev: int
try_: int | None = Field(None, alias="try")
iter: int
rebuilds: List[str]
state: str
branch: str
user: str
runby: str
testonly: int
failearly: int
shared: int
depends: List[int]
message: str
version: str
last_changed: str
subtasks: List[TaskInfoSubtaskModel]
plan: TaskInfoPlan2Model | None
class TaskPackagesPackageElementModel(BaseModel):
name: str
epoch: int
version: str
release: str
disttag: str
buildtime: str
arch: str
class TaskPackagesSubtaskElementModel(BaseModel):
subtask: int
source: TaskPackagesPackageElementModel | None
binaries: List[TaskPackagesPackageElementModel]
class TaskPackagesModel(BaseModel):
model_config = {"populate_by_name": True}
id: int
repo: str
owner: str
state: str
testonly: int
try_: int | None = Field(None, alias="try")
iter: int
message: str
dependencies: List[int]
length: int
subtasks: List[TaskPackagesSubtaskElementModel]
arepo: List[TaskPackagesPackageElementModel]
class TaskDiffDependenciesModel(BaseModel):
model_config = {"populate_by_name": True}
type: str
del_: List[str] = Field(default_factory=list, alias="del")
add: List[str]
class TaskDiffPackagesModel(BaseModel):
model_config = {"populate_by_name": True}
package: str
del_: List[str] = Field(default_factory=list, alias="del")
add: List[str]
dependencies: List[TaskDiffDependenciesModel]
class TaskDiffArchsModel(BaseModel):
arch: str
packages: List[TaskDiffPackagesModel]
class TaskDiffModel(BaseModel):
task_id: int
task_have_plan: bool
task_diff: List[TaskDiffArchsModel]
class TaskFindPackagesetElementModel(BaseModel):
branch: str
pkgset_datetime: str
sourcepkgname: str
packages: List[str]
version: str
release: str
disttag: str
packager_email: str
buildtime: str
archs: List[str]
class TaskFindPackagesetModel(BaseModel):
id: int
request_args: Dict[str, Any]
task_packages: List[str]
length: int
packages: List[TaskFindPackagesetElementModel]
class TaskMisconflictPackageModel(BaseModel):
input_package: str
conflict_package: str
version: str
release: str
epoch: int
archs: List[str]
files_with_conflict: List[str]
class TaskMisconflictPackagesModel(BaseModel):
id: int
request_args: Dict[str, Any]
length: int
conflicts: List[TaskMisconflictPackageModel]
class NeedsApprovalSubtaskElementModel(BaseModel):
id: int
type: str
package: str
userid: str
dir: str
sid: str
pkg_from: str
tag_author: str
tag_id: str
tag_name: str
srpm: str
srpm_name: str
srpm_evr: str
last_changed: str
source_package: TaskInfoPackageModel | None
class NeedsApprovalTaskElementModel(BaseModel):
model_config = {"populate_by_name": True}
id: int
state: str
runby: str
try_: int | None = Field(None, alias="try")
iter: int
failearly: bool
shared: bool
depends: List[int]
testonly: bool
message: str
version: str
prev: int
last_changed: str
branch: str
user: str
subtasks: List[NeedsApprovalSubtaskElementModel]
class NeedsApprovalModel(BaseModel):
length: int
tasks: List[NeedsApprovalTaskElementModel]
class TaskHistoryElementModel(BaseModel):
task_id: int
task_commited: str
branch_commited: str
class TaskHistoryModel(BaseModel):
request_args: Dict[str, Any]
length: int
tasks: List[TaskHistoryElementModel]
class TaskRepoPackageModel(BaseModel):
name: str
version: str
release: str
filename: str
class TaskRepoInfoModel(BaseModel):
name: str
date: str
tag: str
class TaskRepoArchsModel(BaseModel):
arch: str
packages: List[TaskRepoPackageModel]
class TaskRepoModel(BaseModel):
task_id: int
base_repository: TaskRepoInfoModel | None
task_diff_list: List[int]
archs: List[TaskRepoArchsModel]
class TaskBuildDependencyElementModel(BaseModel):
depth: int
name: str
version: str
release: str
epoch: int
serial: int
sourcerpm: str
branch: str
buildtime: str
archs: List[str]
cycle: List[str]
requires: List[str]
acl: List[str]
class TaskBuildDependencyModel(BaseModel):
id: int
request_args: Dict[str, Any]
length: int
dependencies: List[TaskBuildDependencyElementModel]
class VulnerabilityReferenceElementModel(BaseModel):
name: str
url: str
tags: List[str]
class VulnerabilityCVSSVectorElementModel(BaseModel):
version: str
score: float
vector: str
class VulnerabilityConfigurationElementModel(BaseModel):
cpe: str
version_start_excluding: str
version_start_including: str
version_end_excluding: str
version_end_including: str
class VulnerabilityParsedDetailsModel(BaseModel):
references: List[VulnerabilityReferenceElementModel]
cvss_vectors: List[VulnerabilityCVSSVectorElementModel]
configurations: List[VulnerabilityConfigurationElementModel]
cwes: List[str]
class VulnerabilityModel(BaseModel):
model_config = {"populate_by_name": True}
id: str
summary: str
url: str
severity: str
score: float
published: str
modified: str
refs: List[str]
json_: Dict[str, Any] = Field(default_factory=dict, alias="json")
rejected: bool
class VulnerabilityInfoModel(BaseModel):
request_args: Dict[str, Any]
vuln_info: VulnerabilityModel | None
parsed: VulnerabilityParsedDetailsModel | None
class VulnPackageLastVersionModel(BaseModel):
pkghash: str
name: str
branch: str
version: str
release: str
class VulnFixesPackagesElementModel(BaseModel):
pkghash: str
name: str
branch: str
version: str
release: str
errata_id: str
task_id: int
task_state: str
last_version: VulnPackageLastVersionModel | None
class VulnFixesPackagesModel(BaseModel):
request_args: Dict[str, Any]
length: int
packages: List[VulnFixesPackagesElementModel]
class CveTaskPackageVulnerableElementModel(BaseModel):
id: str
type: str
link: str
class CveTaskPackagesElementModel(BaseModel):
subtask: int
pkghash: str
pkg_name: str
pkg_version: str
pkg_release: str
branch: str
errata_id: str
errata_link: str
vulnerabilities: List[CveTaskPackageVulnerableElementModel]
class CveVulnerableTaskModel(BaseModel):
packages: List[CveTaskPackagesElementModel]
from .methods import ALTRepoAppStream
import os
import aiohttp
from lxml import etree
class AppStreamClient:
def __init__(self, session: aiohttp.ClientSession, appstream_url: str):
self.session = session
self._appstream_url = appstream_url
async def _handle_response(self, resp: aiohttp.ClientResponse):
match resp.status:
case 200 | 201:
return await resp.text(encoding="utf-8")
case _:
return None
async def get(self, branch: str) -> str | None:
url = self._appstream_url.format(branch=branch)
async with self.session.get(url) as resp:
return await self._handle_response(resp)
class DataInfo:
def __init__(self, client: AppStreamClient, appstream_dir: str):
self.client = client
self.dir = appstream_dir
self._on_update = None
os.makedirs(self.dir, exist_ok=True)
async def load_by_branch(self, branch: str, version: str):
text = await self.client.get(branch)
if not text:
return
old_file = self.get_file_path(branch)
if old_file:
os.remove(old_file)
filename = f"{branch}-{version}.xml"
filepath = os.path.join(self.dir, filename)
with open(filepath, "w", encoding="utf-8") as f:
f.write(text)
if self._on_update:
self._on_update(branch)
def get_file_path(self, branch: str) -> str | None:
for fname in os.listdir(self.dir):
if fname.startswith(f"{branch}-") and fname.endswith(".xml"):
return os.path.join(self.dir, fname)
return None
def get_current_version(self, branch: str) -> str | None:
path = self.get_file_path(branch)
if path:
fname = os.path.basename(path)
return fname[len(branch) + 1 : -4]
return None
def has_file(self, branch: str) -> bool:
return self.get_file_path(branch) is not None
class PackageInfo:
def __init__(self, client: AppStreamClient, data: DataInfo):
self.client = client
self._data = data
self._cache: dict[str, dict[str, str]] = {}
def rebuild_cache(self, branch: str) -> None:
path = self._data.get_file_path(branch)
if not path:
self._cache.pop(branch, None)
return
result: dict[str, str] = {}
context = etree.iterparse(path, events=("end",), recover=True)
for _, elem in context:
if elem.tag == "component" and elem.get("type") != "addon":
pkgname = elem.findtext("pkgname")
app_id = elem.findtext("id")
if pkgname and app_id:
result[pkgname] = app_id
elem.clear()
self._cache[branch] = result
def id_by_pkgname(self, pkgname: str, branch: str) -> str | None:
if branch not in self._cache:
return None
return self._cache[branch].get(pkgname)
class ALTRepoAppStream:
def __init__(self, session: aiohttp.ClientSession, config: "ALTRepoConfig"):
self.branches = config.appstream_branches
self._client = AppStreamClient(session, config.appstream_url)
self.data = DataInfo(self._client, config.appstream_dir)
self.package = PackageInfo(self._client, self.data)
self.data._on_update = self.package.rebuild_cache
for branch in self.branches:
if self.data.has_file(branch):
self.package.rebuild_cache(branch)
from dataclasses import dataclass, field
@dataclass
class ALTRepoConfig:
api_base_url: str = "https://rdb.altlinux.org/api"
cybertalk_url: str = "https://lists.altlinux.org/pipermail/sisyphus-cybertalk/{}/"
ftbfs_url: str = "https://git.altlinux.org/beehive/stats/Sisyphus-x86_64/ftbfs-joined"
watch_url: str = "https://watch.altlinux.org/pub/watch/{by_acl}/{nickname}.txt"
appstream_url: str = (
"https://git.altlinux.org/gears/a/appstream-data-desktop.git?"
"a=blob_plain;f=xmls/altlinux.xml;hb=refs/heads/{branch}"
)
appstream_branches: list[str] = field(default_factory=lambda: ["sisyphus", "p11"])
appstream_dir: str = "appstream"
from .methods import ALTRepoParser
import aiohttp
from datetime import date
from typing import List, Literal
from . import models
from .news import urls_parser, urls_for_range, packages_parser, bugs_parser
from .packages import ftbfs_parser, watch_parser
class BaseParser:
def __init__(self, session: aiohttp.ClientSession):
self.session = session
async def get(self, url: str, encoding: str = "utf-8"):
async with self.session.get(url) as resp:
resp.raise_for_status()
return await resp.text(encoding=encoding)
async def get_bytes(self, url: str) -> bytes:
async with self.session.get(url) as resp:
resp.raise_for_status()
return await resp.read()
class NewsInfo:
def __init__(self, client: BaseParser, cybertalk_url: str):
self.client = client
self._cybertalk_url = cybertalk_url
async def news_urls(self) -> models.NewsURL:
return await urls_parser(self.client, self._cybertalk_url)
async def _get_packages(self, branch: str) -> models.PackagesModel | None:
url = getattr(await self.news_urls(), branch, None)
if not url:
return None
html = await self.client.get(url, "koi8-r")
return await packages_parser(html, url, self.client)
async def bugs(self) -> models.BugsModel | None:
url = (await self.news_urls()).bugs
if not url:
return None
html = await self.client.get(url, "koi8-r")
return await bugs_parser(html, url)
async def sisyphus(self) -> models.PackagesModel | None:
return await self._get_packages("sisyphus")
async def p11(self) -> models.PackagesModel | None:
return await self._get_packages("p11")
async def p10(self) -> models.PackagesModel | None:
return await self._get_packages("p10")
async def bugs_by_range(
self, date_from: date, date_to: date
) -> dict[str, int] | None:
urls = await urls_for_range(self.client, date_from, date_to, self._cybertalk_url, "bugs")
if not urls:
return None
totals = {
"quickly_resolved": 0, "new": 0, "old": 0,
"resolved": 0, "reopened": 0, "random": 0,
}
for _, url in urls:
html = await self.client.get(url, "koi8-r")
data = await bugs_parser(html, url)
if data and isinstance(data, models.BugsModel):
for key in totals:
items = getattr(data, key, None)
if items:
totals[key] += len(items)
if all(v == 0 for v in totals.values()):
return None
return totals
async def packages_by_range(
self, date_from: date, date_to: date
) -> models.PackagesModel | None:
urls = await urls_for_range(self.client, date_from, date_to, self._cybertalk_url)
if not urls:
return None
all_packages = []
for _, url in urls:
html = await self.client.get(url, "koi8-r")
packages = await packages_parser(html, url, self.client)
if packages and isinstance(packages, models.PackagesModel):
all_packages.append(packages)
if not all_packages:
return None
if len(all_packages) == 1:
return all_packages[0]
return _aggregate_packages(all_packages)
def _aggregate_packages(
packages_list: list[models.PackagesModel],
) -> models.PackagesModel:
latest_added = {}
latest_removed = {}
latest_updated = {}
for packages in packages_list:
for pkg in (packages.added or []):
latest_added[pkg.name] = pkg
for pkg in (packages.removed or []):
latest_removed[pkg.name] = pkg
for pkg in (packages.updated or []):
latest_updated[pkg.name] = pkg
# Добавлен + удалён → убираем из обоих
cancelled = set(latest_added) & set(latest_removed)
for name in cancelled:
del latest_added[name]
del latest_removed[name]
# Добавлен + обновлён → оставляем в "добавлено" с последними данными
for name in list(latest_updated):
if name in latest_added:
latest_added[name] = latest_updated.pop(name)
elif name in latest_removed:
del latest_updated[name]
added = list(latest_added.values())
removed = list(latest_removed.values())
updated = list(latest_updated.values())
return models.PackagesModel(
url=packages_list[-1].url,
total=packages_list[-1].total,
added=added or None,
removed=removed or None,
updated=updated or None,
)
class PackagesInfo:
def __init__(self, client: BaseParser, ftbfs_url: str, watch_url: str):
self.client = client
self._ftbfs_url = ftbfs_url
self._watch_url = watch_url
async def ftbfs(self) -> List[models.FTBFSModel]:
text = await self.client.get(self._ftbfs_url)
return ftbfs_parser(text)
async def watch_by_maintainer(
self,
maintainer_nickname: str,
by_acl: Literal["by-acl", "by-expanded-acl", "by-expanded-leader", "by-leader"],
) -> List[models.WatchByMaintainerModel]:
url = self._watch_url.format(by_acl=by_acl, nickname=maintainer_nickname)
try:
text = await self.client.get(url)
return watch_parser(text)
except:
return []
class ALTRepoParser:
def __init__(self, session: aiohttp.ClientSession, config: "ALTRepoConfig"):
self._client = BaseParser(session)
self.news = NewsInfo(self._client, config.cybertalk_url)
self.packages = PackagesInfo(self._client, config.ftbfs_url, config.watch_url)
from pydantic import BaseModel
from typing import List
class NewsURL(BaseModel):
sisyphus: str | None = None
p11: str | None = None
p10: str | None = None
bugs: str | None = None
class BugsElementModel(BaseModel):
id: int
component: str
priority: str
status: str | None = None
summary: str
class BugsModel(BaseModel):
url: str
quickly_resolved: List[BugsElementModel] | None = None
new: List[BugsElementModel] | None = None
old: List[BugsElementModel] | None = None
resolved: List[BugsElementModel] | None = None
reopened: List[BugsElementModel] | None = None
random: List[BugsElementModel] | None = None
class RemovedPackageElementModel(BaseModel):
name: str
version: str
class PackageElementModel(BaseModel):
name: str
description: str
maintainer_name: str
maintainer_nick: str
class PackagesModel(BaseModel):
url: str
total: int
added: List[PackageElementModel] | None = None
removed: List[RemovedPackageElementModel] | None = None
updated: List[PackageElementModel] | None = None
class FTBFSModel(BaseModel):
name: str
version: str
ftbfs_weeks: int
maintainers: List[str]
class WatchByMaintainerModel(BaseModel):
pkg_name: str
old_version: str
new_version: str
url: str
from .urls import urls_parser, urls_for_range
from .packages import packages_parser
from .bugs import bugs_parser
from bs4 import BeautifulSoup
import re
from .. import models
async def bugs_parser(html: str, url: str):
soup = BeautifulSoup(html, "html.parser")
pre_tag = soup.find("pre")
if not pre_tag:
return {}
text = pre_tag.get_text()
lines = text.strip().splitlines()
data = {}
section_name = None
bug_pattern = re.compile(r"#(\d+)\s+([^\t]+)\s+([^\t]+)\s+([^\t]+)")
description_buffer = ""
current_bug = None
for line in lines:
header_match = re.match(
r"^\s*(\d+)\s+(NEW|RESOLVED|REOPENED|RANDOM|OLD)\s+bugs?.*", line)
if header_match:
if current_bug and section_name:
current_bug["summary"] = description_buffer.strip()
data.setdefault(section_name, []).append(current_bug)
current_bug = None
description_buffer = ""
section_name = _get_bug_section_name(line)
continue
bug_match = bug_pattern.match(line)
if bug_match:
if current_bug and section_name:
current_bug["summary"] = description_buffer.strip()
data.setdefault(section_name, []).append(current_bug)
status = bug_match.group(4).strip()
if status == "---":
status = None
current_bug = {
"id": int(bug_match.group(1)),
"component": bug_match.group(2).strip(),
"priority": bug_match.group(3).strip(),
"status": status
}
description_buffer = ""
elif current_bug:
if not line.strip():
continue
if re.match(r"^Total\s+\d+\s+pending bugs", line.strip(), re.IGNORECASE):
break
description_buffer += line.strip() + " "
if current_bug and section_name:
current_bug["summary"] = description_buffer.strip()
data.setdefault(section_name, []).append(current_bug)
data["url"] = url
return models.BugsModel(**data)
def _get_bug_section_name(line: str) -> str:
line = line.lower()
if "new" in line and "resolved" in line:
return "quickly_resolved"
elif "new" in line:
return "new"
elif "old" in line:
return "old"
elif "resolved" in line:
return "resolved"
elif "reopened" in line:
return "reopened"
elif "random" in line:
return "random"
from bs4 import BeautifulSoup
import re
import io
import gzip
from .. import models
async def packages_parser(html: str, url: str, client=None):
soup = BeautifulSoup(html, "html.parser")
pre_tag = soup.find("pre")
if not pre_tag:
return {}
pre_text = pre_tag.get_text(strip=True)
if "Было удалено вложение" in pre_text and "attachment" in pre_text:
attachment_link = pre_tag.find("a", href=True)
if attachment_link and client:
attachment_url = attachment_link["href"]
compressed_data = await client.get_bytes(attachment_url)
with gzip.GzipFile(fileobj=io.BytesIO(compressed_data)) as gz:
text = gz.read().decode('utf-8')
else:
return models.PackagesModel(**{"added": [], "removed": [], "updated": [], "url": "none"})
else:
text = pre_tag.get_text()
lines = text.strip().splitlines()
sections = {"added": [], "removed": [], "updated": []}
current_section = None
current_package = {}
seen_changelog = False
for line in lines:
# Смена секции
if re.search(r"^\s*\d+\s+ADDED package[s]?", line):
if current_package and current_section:
sections[current_section].append(current_package)
current_section = "added"
current_package = {}
seen_changelog = False
continue
elif re.search(r"^\s*\d+\s+REMOVED package[s]?", line):
if current_package and current_section:
sections[current_section].append(current_package)
current_section = "removed"
current_package = {}
seen_changelog = False
continue
elif re.search(r"^\s*\d+\s+UPDATED packages[s]?", line):
if current_package and current_section:
sections[current_section].append(current_package)
current_section = "updated"
current_package = {}
seen_changelog = False
continue
if current_section == "removed":
if line.strip():
parts = line.strip().split("\t")
if len(parts) == 2:
name, version = parts
sections["removed"].append({
"name": name.strip(),
"version": version.strip()
})
continue
match = re.match(r"^(\S+)\s+-\s+(.*)", line)
if match and not line.startswith("- "):
if current_package and current_section:
sections[current_section].append(current_package)
current_package = {
"name": match.group(1),
"description": _clean_description(match.group(2)),
}
seen_changelog = False
continue
if (
current_package
and current_section != "removed"
and not seen_changelog
):
match = re.match(r"^\*\s+\w+\s+\w+\s+\d+\s+\d+\s+(.+?) <([^@\s>]+)(?:@altlinux| на altlinux)>", line)
if match:
current_package["maintainer_name"] = match.group(1).strip()
current_package["maintainer_nick"] = match.group(2).strip()
seen_changelog = True
continue
match = re.search(r'^Total (\d+) source packages\.$', line)
if match:
sections["total"] = int(match.group(1))
if current_package and current_section:
sections[current_section].append(current_package)
sections["url"] = url
return models.PackagesModel(**sections)
def _clean_description(desc: str):
desc = desc.strip()
desc = re.sub(r'\s+', ' ', desc)
desc = re.sub(r'\[\d+[KMG]?\]', '', desc).strip()
return desc
from bs4 import BeautifulSoup
import re
from datetime import datetime, date
from .. import models
async def urls_parser(client, cybertalk_url: str):
now = datetime.now()
year_month = f"{now.year}-{now.strftime("%B")}"
today = now.strftime("%Y%m%d")
base_url = cybertalk_url.format(year_month)
html = await client.get(f"{base_url}date.html", "koi8-r")
soup = BeautifulSoup(html, "html.parser")
result = {"sisyphus": None, "bugs": None, "p11": None}
for li in soup.find_all("li"):
a = li.find("a")
if not a or not a.get("href"):
continue
href = a["href"]
text = a.get_text(strip=True)
url = base_url + href
if f"Sisyphus-{today} packages" in text:
result["sisyphus"] = url
elif f"Sisyphus-{today} bugs" in text:
result["bugs"] = url
elif "p11/branch packages" in text and await _check_date(url, client):
result["p11"] = url
elif "p10/branch packages" in text and await _check_date(url, client):
result["p10"] = url
return models.NewsURL(**result)
async def urls_for_range(
client, date_from: date, date_to: date,
cybertalk_url: str, news_type: str = "packages",
) -> list[tuple[date, str]]:
results = []
pattern = re.compile(rf"Sisyphus-(\d{{8}}) {re.escape(news_type)}")
current = date_from.replace(day=1)
while current <= date_to:
year_month = f"{current.year}-{current.strftime('%B')}"
base_url = cybertalk_url.format(year_month)
try:
html = await client.get(f"{base_url}date.html", "koi8-r")
except:
current = _next_month(current)
continue
soup = BeautifulSoup(html, "html.parser")
for li in soup.find_all("li"):
a = li.find("a")
if not a or not a.get("href"):
continue
text = a.get_text(strip=True)
match = pattern.search(text)
if match:
news_date = datetime.strptime(match.group(1), "%Y%m%d").date()
if date_from <= news_date <= date_to:
results.append((news_date, base_url + a["href"]))
current = _next_month(current)
return sorted(results)
def _next_month(d: date) -> date:
if d.month == 12:
return d.replace(year=d.year + 1, month=1)
return d.replace(month=d.month + 1)
async def _check_date(url, client):
html = await client.get(url, "koi8-r")
soup = BeautifulSoup(html, "html.parser")
tag = soup.find("i")
if not tag:
return False
parts = tag.text.strip().split()
if len(parts) < 3:
return False
try:
return int(parts[2]) == datetime.now().day
except ValueError:
return False
from .ftbfs import ftbfs_parser
from .watch import watch_parser
from .. import models
def ftbfs_parser(text: str):
packages = []
for line in text.strip().splitlines():
parts = line.split('\t')
if len(parts) != 4:
continue
name, version, weeks, maintainers = parts
packages.append(models.FTBFSModel(
name = name,
version = version,
ftbfs_weeks = int(weeks),
maintainers = maintainers.split(','),
))
return packages
from .. import models
def watch_parser(text: str):
return [
models.WatchByMaintainerModel(
pkg_name=parts[0],
old_version=parts[1],
new_version=parts[2],
url=parts[3],
)
for parts in (line.split("\t") for line in text.strip().splitlines())
if len(parts) == 4
]
[project]
name = "altrepo"
version = "0.1.0"
description = "Async client for ALT Linux repository services (rdb API, packages parser, AppStream)"
authors = [
{name = "Kirill Unitsaev",email = "fiersik@altlinux.org"}
]
license = {text = "AGPL-3.0-or-later"}
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"aiohttp>=3.9.0",
"pydantic>=2.0.0",
"beautifulsoup4>=4.12.0",
"lxml>=5.0.0",
]
[tool.poetry]
packages = [{include = "altrepo"}]
[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
build-backend = "poetry.core.masonry.api"
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment