feat(nicegui): show list of all images stored in container registry (#43)

This commit is contained in:
Stephan Lüscher 2024-05-15 20:39:20 +00:00
parent b40e0dd4a1
commit 83f047f2d6
No known key found for this signature in database
GPG key ID: 445779060FF3D3CF
8 changed files with 417 additions and 10 deletions

View file

@ -1,10 +1,38 @@
import pandas
import humanize
from nicegui import ui
import os
from utils.registry import DockerRegistry
## TODO: this should be async but I currently don't know how to implement this without button press
def get_image_info() -> pandas.DataFrame:
registry = DockerRegistry()
all_image_info = registry.get_all_image_info()
if isinstance(all_image_info, list) and len(all_image_info) > 0:
data = pandas.json_normalize(
all_image_info,
record_path=["tags"],
meta=[["name"]],
meta_prefix="image_",
).assign(
size=lambda x: x["manifest.layers"].apply(
lambda layers: sum(layer["size"] for layer in layers)
)
)
data = data[["image_name", "name", "size"]].rename(
columns={"image_name": "image", "name": "tag", "size": "size"}
)
data["size"] = data["size"].apply(humanize.naturalsize)
return data
else:
ui.notify(message="No images found")
data = pandas.DataFrame(columns=["image_name", "tag", "size"])
return data
def content() -> None:
project_root = os.environ['NICEGUI_DIR']
ui.label("Work in progress...").classes("text-h6")
ui.image(project_root + "/pages/assets/work-in-progress.png").classes(
"w-[200%]"
)
with ui.row().classes("w-full"):
with ui.card().classes("w-full"):
ui.label("Image Overview").classes("text-h5")
data = get_image_info()
ui.table.from_pandas(df=data).classes("w-full")

View file

@ -0,0 +1,65 @@
import requests
from nicegui import ui
class DockerRegistry:
def __init__(
self,
*,
registry_url: str = "https://registry.ublue.local",
registry_root_cert: str = "/data/ublue-os_forge-root.pem",
) -> None:
"""
Docker Registry integration for ublue-os forge
"""
self.registry_url = registry_url
self.registry_root_cert = registry_root_cert
def get_repositories(self) -> list:
endpoint = f"{self.registry_url}/v2/_catalog"
response = requests.get(url=endpoint, verify=self.registry_root_cert)
if response.status_code == 200:
repositories = response.json()["repositories"]
else:
with self, ui.notify() as notify:
notify(message=f"Error: {response.text}")
return repositories
def get_image_tags(self, image_name=str) -> dict:
endpoint = f"{self.registry_url}/v2/{image_name}/tags/list"
response = requests.get(url=endpoint, verify=self.registry_root_cert)
if response.status_code == 200:
tags = response.json()["tags"]
else:
with self, ui.notify() as notify:
notify(message=f"Error: {response.text}")
return tags
def get_image_manifest(self, image_name=str, image_tag=str) -> dict:
endpoint = f"{self.registry_url}/v2/{image_name}/manifests/{image_tag}"
headers = {"accept": "application/vnd.oci.image.manifest.v1+json"}
response = requests.get(
url=endpoint, verify=self.registry_root_cert, headers=headers
)
if response.status_code == 200:
manifest = response.json()
else:
with self, ui.notify() as notify:
notify(message=f"Error: {response.text}")
return manifest
def get_all_image_info(self, image_name=str) -> list:
repositories = self.get_repositories()
all_image_info = []
for repository in repositories:
tags = self.get_image_tags(image_name=repository)
for tag in tags:
manifest = self.get_image_manifest(image_name=repository, image_tag=tag)
image_info = {
"name": repository,
"tags": [{"name": tag, "manifest": manifest}],
}
all_image_info.append(image_info)
return all_image_info