api: refactor nested URL methods into subclasses with deprecated aliases

parent a57b6974
import aiohttp import aiohttp
from urllib.parse import urlencode from urllib.parse import urlencode
from warnings import deprecated
from typing import List, Literal from typing import List, Literal
...@@ -1321,9 +1322,42 @@ class DependenciesInfo: ...@@ -1321,9 +1322,42 @@ class DependenciesInfo:
return models.DependsPackageBuildDependencyModel(**data) return models.DependsPackageBuildDependencyModel(**data)
class ErrataExportOvalInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def branches(self) -> models.OvalBranchesModel:
data = await self.client.get("/errata/export/oval/branches")
return models.OvalBranchesModel(**data)
async def info(
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
)
class ErrataExportInfo:
def __init__(self, client: BaseAPI):
self.oval = ErrataExportOvalInfo(client)
class ErrataInfo: class ErrataInfo:
def __init__(self, client: BaseAPI): def __init__(self, client: BaseAPI):
self.client = client self.client = client
self.export = ErrataExportInfo(self.client)
async def advisory( async def advisory(
self, self,
...@@ -1359,28 +1393,6 @@ class ErrataInfo: ...@@ -1359,28 +1393,6 @@ class ErrataInfo:
data = await self.client.get("/errata/errata_branches") data = await self.client.get("/errata/errata_branches")
return models.ErrataBranchesModel(**data) 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( async def find_erratas(
self, self,
input: list[str] | None = None, input: list[str] | None = None,
...@@ -1469,12 +1481,20 @@ class ErrataInfo: ...@@ -1469,12 +1481,20 @@ class ErrataInfo:
data = await self.client.get("/errata/search", params or None) data = await self.client.get("/errata/search", params or None)
return models.ErratasModel(**data) return models.ErratasModel(**data)
@deprecated("use errata.export.oval.branches() instead")
async def export_oval_branches(self):
return await self.export.oval.branches()
class ExportInfo: @deprecated("use errata.export.oval.info() instead")
async def export_oval(self, branch, package_name=None, one_file=False):
return await self.export.oval.info(branch, package_name, one_file)
class ExportBeehiveInfo:
def __init__(self, client: BaseAPI): def __init__(self, client: BaseAPI):
self.client = client self.client = client
async def beehive_ftbfs( async def ftbfs(
self, branch: str, arch: str | None = None self, branch: str, arch: str | None = None
) -> models.ExportBeehiveFTBFSListModel: ) -> models.ExportBeehiveFTBFSListModel:
params = { params = {
...@@ -1488,6 +1508,33 @@ class ExportInfo: ...@@ -1488,6 +1508,33 @@ class ExportInfo:
data = await self.client.get("/export/beehive/ftbfs", params) data = await self.client.get("/export/beehive/ftbfs", params)
return models.ExportBeehiveFTBFSListModel(**data) return models.ExportBeehiveFTBFSListModel(**data)
class ExportTranslationInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def 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 ExportInfo:
def __init__(self, client: BaseAPI):
self.client = client
self.beehive = ExportBeehiveInfo(self.client)
self.translation = ExportTranslationInfo(self.client)
async def branch_binary_packages( async def branch_binary_packages(
self, branch: str, arch: str | None = None self, branch: str, arch: str | None = None
) -> models.PackagesetPackagesExportModel: ) -> models.PackagesetPackagesExportModel:
...@@ -1515,25 +1562,44 @@ class ExportInfo: ...@@ -1515,25 +1562,44 @@ class ExportInfo:
data = await self.client.get(f"/export/sitemap_packages/{branch}") data = await self.client.get(f"/export/sitemap_packages/{branch}")
return models.SitemapPackagesExportModel(**data) return models.SitemapPackagesExportModel(**data)
async def translation_packages_po_files( @deprecated("use export.beehive.ftbfs() instead")
self, branches: list[str], from_date: str | None = None async def beehive_ftbfs(self, branch, arch=None):
) -> bytes: return await self.beehive.ftbfs(branch, arch)
params = {
k: v @deprecated("use export.translation.packages_po_files() instead")
for k, v in { async def translation_packages_po_files(self, branches, from_date=None):
"branches": branches, return await self.translation.packages_po_files(branches, from_date)
"from_date": from_date,
}.items()
if v is not None class ImageInspectInfo:
} def __init__(self, client: BaseAPI):
return await self.client.get_bytes( self.client = client
"/export/translation/packages_po_files", params
) async def regular(
self, payload: dict
) -> models.ImagePackagesInspectRegularModel:
data = await self.client.post("/image/inspect/regular", json=payload)
return models.ImagePackagesInspectRegularModel(**data)
async def sp(self, payload: dict) -> models.ImagePackagesInspectSPModel:
data = await self.client.post("/image/inspect/sp", json=payload)
return models.ImagePackagesInspectSPModel(**data)
class ImageIsoInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def all_images(self) -> models.ImageAllISOModel:
data = await self.client.get("/image/iso/all_images")
return models.ImageAllISOModel(**data)
class ImageInfo: class ImageInfo:
def __init__(self, client: BaseAPI): def __init__(self, client: BaseAPI):
self.client = client self.client = client
self.inspect = ImageInspectInfo(self.client)
self.iso = ImageIsoInfo(self.client)
async def active_images( async def active_images(
self, self,
...@@ -1674,20 +1740,6 @@ class ImageInfo: ...@@ -1674,20 +1740,6 @@ class ImageInfo:
data = await self.client.get("/image/image_uuid_by_tag", {"tag": tag}) data = await self.client.get("/image/image_uuid_by_tag", {"tag": tag})
return models.ImageTagUUIDModel(**data) 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( async def last_packages_by_image(
self, self,
branch: str, branch: str,
...@@ -1730,6 +1782,18 @@ class ImageInfo: ...@@ -1730,6 +1782,18 @@ class ImageInfo:
) )
return models.LastImagePackagesModel(**data) return models.LastImagePackagesModel(**data)
@deprecated("use image.inspect.regular() instead")
async def inspect_regular(self, payload):
return await self.inspect.regular(payload)
@deprecated("use image.inspect.sp() instead")
async def inspect_sp(self, payload):
return await self.inspect.sp(payload)
@deprecated("use image.iso.all_images() instead")
async def iso_all_images(self):
return await self.iso.all_images()
class LicenseInfo: class LicenseInfo:
def __init__(self, client: BaseAPI): def __init__(self, client: BaseAPI):
...@@ -1747,51 +1811,61 @@ class LicenseInfo: ...@@ -1747,51 +1811,61 @@ class LicenseInfo:
return models.LicenseTokensModel(**data) return models.LicenseTokensModel(**data)
class VulnInfo: class CveInfo:
def __init__(self, client: BaseAPI): def __init__(self, client: BaseAPI):
self.client = client self.client = client
async def bdu( async def info(
self, vuln_id: str, exclude_json: bool = False self, vuln_id: str, exclude_json: bool = False
) -> models.VulnerabilityInfoModel: ) -> models.VulnerabilityInfoModel:
data = await self.client.get( data = await self.client.get(
"/vuln/bdu", {"vuln_id": vuln_id, "exclude_json": exclude_json} "/vuln/cve", {"vuln_id": vuln_id, "exclude_json": exclude_json}
) )
return models.VulnerabilityInfoModel(**data) return models.VulnerabilityInfoModel(**data)
async def bdu_fixes( async def fixes(
self, vuln_id: str, exclude_json: bool = False self, vuln_id: str, exclude_json: bool = False
) -> models.VulnFixesPackagesModel: ) -> models.VulnFixesPackagesModel:
data = await self.client.get( data = await self.client.get(
"/vuln/bdu/fixes", {"vuln_id": vuln_id, "exclude_json": exclude_json} "/vuln/cve/fixes", {"vuln_id": vuln_id, "exclude_json": exclude_json}
) )
return models.VulnFixesPackagesModel(**data) return models.VulnFixesPackagesModel(**data)
async def cve( async def excluded(
self, vuln_id: str, exclude_json: bool = False self, vuln_id: str, exclude_json: bool = False
) -> models.VulnerabilityInfoModel: ) -> models.VulnFixesPackagesModel:
data = await self.client.get( data = await self.client.get(
"/vuln/cve", {"vuln_id": vuln_id, "exclude_json": exclude_json} "/vuln/cve/excluded", {"vuln_id": vuln_id, "exclude_json": exclude_json}
) )
return models.VulnerabilityInfoModel(**data) return models.VulnFixesPackagesModel(**data)
class BduInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def cve_excluded( async def info(
self, vuln_id: str, exclude_json: bool = False self, vuln_id: str, exclude_json: bool = False
) -> models.VulnFixesPackagesModel: ) -> models.VulnerabilityInfoModel:
data = await self.client.get( data = await self.client.get(
"/vuln/cve/excluded", {"vuln_id": vuln_id, "exclude_json": exclude_json} "/vuln/bdu", {"vuln_id": vuln_id, "exclude_json": exclude_json}
) )
return models.VulnFixesPackagesModel(**data) return models.VulnerabilityInfoModel(**data)
async def cve_fixes( async def fixes(
self, vuln_id: str, exclude_json: bool = False self, vuln_id: str, exclude_json: bool = False
) -> models.VulnFixesPackagesModel: ) -> models.VulnFixesPackagesModel:
data = await self.client.get( data = await self.client.get(
"/vuln/cve/fixes", {"vuln_id": vuln_id, "exclude_json": exclude_json} "/vuln/bdu/fixes", {"vuln_id": vuln_id, "exclude_json": exclude_json}
) )
return models.VulnFixesPackagesModel(**data) return models.VulnFixesPackagesModel(**data)
async def ghsa(
class GhsaInfo:
def __init__(self, client: BaseAPI):
self.client = client
async def info(
self, vuln_id: str, exclude_json: bool = False self, vuln_id: str, exclude_json: bool = False
) -> models.VulnerabilityInfoModel: ) -> models.VulnerabilityInfoModel:
data = await self.client.get( data = await self.client.get(
...@@ -1799,7 +1873,7 @@ class VulnInfo: ...@@ -1799,7 +1873,7 @@ class VulnInfo:
) )
return models.VulnerabilityInfoModel(**data) return models.VulnerabilityInfoModel(**data)
async def ghsa_fixes( async def fixes(
self, vuln_id: str, exclude_json: bool = False self, vuln_id: str, exclude_json: bool = False
) -> models.VulnFixesPackagesModel: ) -> models.VulnFixesPackagesModel:
data = await self.client.get( data = await self.client.get(
...@@ -1807,10 +1881,46 @@ class VulnInfo: ...@@ -1807,10 +1881,46 @@ class VulnInfo:
) )
return models.VulnFixesPackagesModel(**data) return models.VulnFixesPackagesModel(**data)
class VulnInfo:
def __init__(self, client: BaseAPI):
self.client = client
self.cve = CveInfo(self.client)
self.bdu = BduInfo(self.client)
self.ghsa = GhsaInfo(self.client)
async def task(self, task_id: int) -> models.CveVulnerableTaskModel: async def task(self, task_id: int) -> models.CveVulnerableTaskModel:
data = await self.client.get(f"/vuln/task/{task_id}") data = await self.client.get(f"/vuln/task/{task_id}")
return models.CveVulnerableTaskModel(**data) return models.CveVulnerableTaskModel(**data)
@deprecated("use vuln.cve.info() instead")
async def cve(self, vuln_id: str, exclude_json: bool = False):
return await self.cve.info(vuln_id, exclude_json)
@deprecated("use vuln.cve.fixes() instead")
async def cve_fixes(self, vuln_id: str, exclude_json: bool = False):
return await self.cve.fixes(vuln_id, exclude_json)
@deprecated("use vuln.cve.excluded() instead")
async def cve_excluded(self, vuln_id: str, exclude_json: bool = False):
return await self.cve.excluded(vuln_id, exclude_json)
@deprecated("use vuln.bdu.info() instead")
async def bdu(self, vuln_id: str, exclude_json: bool = False):
return await self.bdu.info(vuln_id, exclude_json)
@deprecated("use vuln.bdu.fixes() instead")
async def bdu_fixes(self, vuln_id: str, exclude_json: bool = False):
return await self.bdu.fixes(vuln_id, exclude_json)
@deprecated("use vuln.ghsa.info() instead")
async def ghsa(self, vuln_id: str, exclude_json: bool = False):
return await self.ghsa.info(vuln_id, exclude_json)
@deprecated("use vuln.ghsa.fixes() instead")
async def ghsa_fixes(self, vuln_id: str, exclude_json: bool = False):
return await self.ghsa.fixes(vuln_id, exclude_json)
class ALTRepoAPI: class ALTRepoAPI:
def __init__(self, session: aiohttp.ClientSession, config: "ALTRepoConfig"): def __init__(self, session: aiohttp.ClientSession, config: "ALTRepoConfig"):
......
...@@ -8,7 +8,7 @@ async def main(): ...@@ -8,7 +8,7 @@ async def main():
client = ALTRepo() client = ALTRepo()
await client.init() await client.init()
images = await client.api.image.iso_all_images() images = await client.api.image.iso.all_images()
print(f"Всего ISO-образов: {len(images.images)}") print(f"Всего ISO-образов: {len(images.images)}")
for img in images.images[:5]: for img in images.images[:5]:
print(f" {img.name} ({img.branch}, {img.date})") print(f" {img.name} ({img.branch}, {img.date})")
......
...@@ -8,7 +8,7 @@ async def main(): ...@@ -8,7 +8,7 @@ async def main():
client = ALTRepo() client = ALTRepo()
await client.init() await client.init()
fixes = await client.api.vuln.cve_fixes("CVE-2024-6387") fixes = await client.api.vuln.cve.fixes("CVE-2024-6387")
print(f"CVE-2024-6387 — исправлено в {len(fixes.packages)} пакетах:") print(f"CVE-2024-6387 — исправлено в {len(fixes.packages)} пакетах:")
for pkg in fixes.packages[:5]: for pkg in fixes.packages[:5]:
print(f" {pkg.name} {pkg.version}-{pkg.release} ({pkg.branch})") print(f" {pkg.name} {pkg.version}-{pkg.release} ({pkg.branch})")
......
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