harbor_paginate.py

harbor_paginate.py · 6.0 KiB · Python Raw

import asyncio
from typing import Any, List

from harborapi import HarborAsyncClient

class AccumulatingBuffer:
    def __init__(self, capacity: int):
        if capacity <= 0:
            raise ValueError("Capacity must be positive")
        self.capacity = capacity
        self.buf = [None] * capacity
        self.count = 0  # Фактически заполненные слоты

def load(self, elements: List[Any]) -> None:
        """Добавляет элементы: в свободное место - в конец, при полном - перезапись с начала"""
        L = len(elements)
        if L > self.capacity:
            raise IndexError(f"Elements ({L}) exceed capacity ({self.capacity})")

if self.count + L <= self.capacity:
            # Есть свободное место - добавляем в конец
            for i, elem in enumerate(elements):
                self.buf[self.count + i] = elem
            self.count += L
            # Затираем оставшееся место
            for i in range(self.count, self.capacity):
                self.buf[i] = None
        else:
            # Буфер полон - перезаписываем с начала
            for i, elem in enumerate(elements):
                self.buf[i] = elem
            self.count = L
            # Затираем остаток
            for i in range(L, self.capacity):
                self.buf[i] = None

def get_filled(self) -> List[Any]:
        """Возвращает только непустые элементы"""
        return [x for x in self.buf[: self.count] if x is not None]

def size(self) -> int:
        return self.count

def filled_size(self) -> int:
        return len(self.get_filled())

def free_space(self) -> int:
        return self.capacity - self.count

from itertools import islice

def chunks(lst, n):
    for i in range(0, len(lst), n):
        yield lst[i : i + n]  # Генератор вместо списка

async def get_artifacts(
    client: HarborAsyncClient,
    project_name: str,
    repository_name: str,
    page_number: int,
    page_size: int,
):
    # номер страницы с артефактами в Harbor
    harbor_page_number = 1
    # номер страницы с артефактами в RIM
    rim_page_number = 1
    # выделяем буфер под "страницу" артефактов
    artifacts_page_buffer = AccumulatingBuffer(page_size)
    # начинаем считываение
    while True:
        # считываем из Harbor артефакты и загружаем их в буфер
        artifacts = await client.get_artifacts(
            project_name=project_name,
            repository_name=repository_name,
            page=harbor_page_number,
            page_size=10,
            limit=10,
        )

# нет артефактов
        if not artifacts:
            # если страница была неполной, то возвращаем результат
            if rim_page_number == page_number:
                return artifacts_page_buffer.get_filled()
            break

for artifact in artifacts:
            # массив тэгов
            tags = []
            # если есть тэги
            if artifact.tags:
                # перебираем все ссылки на тэги
                for tag_reference in artifact.tags:
                    # формируем имя
                    artifact_name = (
                        f"{project_name}/{repository_name}:{tag_reference.name}"
                    )
                    # засовываем в список
                    tags.append(artifact_name)
            # print(f"Tags {tags}")
            # print(f"Buffer: {artifacts_page_buffer.get_filled()}")
            # print(
            #     f"RIM page number: {rim_page_number}, length: {len(artifacts_page_buffer.get_filled())}"
            # )
            for portion in chunks(tags, page_size):
                # загружаем порцию в буфер
                artifacts_page_buffer.load(portion)
                # Проверяем после каждой порции
                if (
                    artifacts_page_buffer.filled_size() >= page_size
                    and rim_page_number == page_number
                ):
                    return artifacts_page_buffer.get_filled()
                # Считаем полные страницы RIM
                if artifacts_page_buffer.filled_size() >= page_size:
                    # увеличиваем счетчик страниц в RIM
                    rim_page_number += 1
        # увеличиваем счетчик страниц Harbor
        harbor_page_number += 1
    # не страниц, или указана несуществующая страница
    return []

async def main():
    registry_url = "-"
    registry_login = "-"
    registry_password = "-"
    client = HarborAsyncClient(
        url=registry_url + "/api/v2.0",
        username=registry_login,
        secret=registry_password,
        # ограничение по количеству количества записей о соединениях хранимых клиентом
        max_logs=0,
        verify=False,  # TODO: remove them ?
    )
    rbuf = AccumulatingBuffer(10)
    rbuf.load([])
    print(rbuf.get_filled())
    rbuf.load([1, 2, 3, 4, 5, 6, 7, 8, 9])
    print(rbuf.get_filled())
    rbuf.load([1, 2, 3, 4])
    print(rbuf.get_filled())
    # tests
    # project_name = "sbis-hlha"
    # repository_name = "charts/sbis-docker-dispatcher"
    project_name = "corp-base"
    repository_name = "platform/centos"
    artifacts1 = set(await get_artifacts(client, project_name, repository_name, 1, 6))
    print(artifacts1)
    print(len(artifacts1))
    artifacts2 = set(await get_artifacts(client, project_name, repository_name, 2, 6))
    print(artifacts2)
    print(len(artifacts2))
    print(artifacts1 - artifacts2)

if __name__ == "__main__":
    asyncio.run(main())

1	import asyncio
2	from typing import Any, List
3
4	from harborapi import HarborAsyncClient
5
6
7	class AccumulatingBuffer:
8	def __init__(self, capacity: int):
9	if capacity <= 0:
10	raise ValueError("Capacity must be positive")
11	self.capacity = capacity
12	self.buf = [None] * capacity
13	self.count = 0 # Фактически заполненные слоты
14
15	def load(self, elements: List[Any]) -> None:
16	"""Добавляет элементы: в свободное место - в конец, при полном - перезапись с начала"""
17	L = len(elements)
18	if L > self.capacity:
19	raise IndexError(f"Elements ({L}) exceed capacity ({self.capacity})")
20
21	if self.count + L <= self.capacity:
22	# Есть свободное место - добавляем в конец
23	for i, elem in enumerate(elements):
24	self.buf[self.count + i] = elem
25	self.count += L
26	# Затираем оставшееся место
27	for i in range(self.count, self.capacity):
28	self.buf[i] = None
29	else:
30	# Буфер полон - перезаписываем с начала
31	for i, elem in enumerate(elements):
32	self.buf[i] = elem
33	self.count = L
34	# Затираем остаток
35	for i in range(L, self.capacity):
36	self.buf[i] = None
37
38	def get_filled(self) -> List[Any]:
39	"""Возвращает только непустые элементы"""
40	return [x for x in self.buf[: self.count] if x is not None]
41
42	def size(self) -> int:
43	return self.count
44
45	def filled_size(self) -> int:
46	return len(self.get_filled())
47
48	def free_space(self) -> int:
49	return self.capacity - self.count
50
51
52	from itertools import islice
53
54
55	def chunks(lst, n):
56	for i in range(0, len(lst), n):
57	yield lst[i : i + n] # Генератор вместо списка
58
59
60	async def get_artifacts(
61	client: HarborAsyncClient,
62	project_name: str,
63	repository_name: str,
64	page_number: int,
65	page_size: int,
66	):
67	# номер страницы с артефактами в Harbor
68	harbor_page_number = 1
69	# номер страницы с артефактами в RIM
70	rim_page_number = 1
71	# выделяем буфер под "страницу" артефактов
72	artifacts_page_buffer = AccumulatingBuffer(page_size)
73	# начинаем считываение
74	while True:
75	# считываем из Harbor артефакты и загружаем их в буфер
76	artifacts = await client.get_artifacts(
77	project_name=project_name,
78	repository_name=repository_name,
79	page=harbor_page_number,
80	page_size=10,
81	limit=10,
82	)
83
84	# нет артефактов
85	if not artifacts:
86	# если страница была неполной, то возвращаем результат
87	if rim_page_number == page_number:
88	return artifacts_page_buffer.get_filled()
89	break
90
91	for artifact in artifacts:
92	# массив тэгов
93	tags = []
94	# если есть тэги
95	if artifact.tags:
96	# перебираем все ссылки на тэги
97	for tag_reference in artifact.tags:
98	# формируем имя
99	artifact_name = (
100	f"{project_name}/{repository_name}:{tag_reference.name}"
101	)
102	# засовываем в список
103	tags.append(artifact_name)
104	# print(f"Tags {tags}")
105	# print(f"Buffer: {artifacts_page_buffer.get_filled()}")
106	# print(
107	# f"RIM page number: {rim_page_number}, length: {len(artifacts_page_buffer.get_filled())}"
108	# )
109	for portion in chunks(tags, page_size):
110	# загружаем порцию в буфер
111	artifacts_page_buffer.load(portion)
112	# Проверяем после каждой порции
113	if (
114	artifacts_page_buffer.filled_size() >= page_size
115	and rim_page_number == page_number
116	):
117	return artifacts_page_buffer.get_filled()
118	# Считаем полные страницы RIM
119	if artifacts_page_buffer.filled_size() >= page_size:
120	# увеличиваем счетчик страниц в RIM
121	rim_page_number += 1
122	# увеличиваем счетчик страниц Harbor
123	harbor_page_number += 1
124	# не страниц, или указана несуществующая страница
125	return []
126
127
128	async def main():
129	registry_url = "-"
130	registry_login = "-"
131	registry_password = "-"
132	client = HarborAsyncClient(
133	url=registry_url + "/api/v2.0",
134	username=registry_login,
135	secret=registry_password,
136	# ограничение по количеству количества записей о соединениях хранимых клиентом
137	max_logs=0,
138	verify=False, # TODO: remove them ?
139	)
140	rbuf = AccumulatingBuffer(10)
141	rbuf.load([])
142	print(rbuf.get_filled())
143	rbuf.load([1, 2, 3, 4, 5, 6, 7, 8, 9])
144	print(rbuf.get_filled())
145	rbuf.load([1, 2, 3, 4])
146	print(rbuf.get_filled())
147	# tests
148	# project_name = "sbis-hlha"
149	# repository_name = "charts/sbis-docker-dispatcher"
150	project_name = "corp-base"
151	repository_name = "platform/centos"
152	artifacts1 = set(await get_artifacts(client, project_name, repository_name, 1, 6))
153	print(artifacts1)
154	print(len(artifacts1))
155	artifacts2 = set(await get_artifacts(client, project_name, repository_name, 2, 6))
156	print(artifacts2)
157	print(len(artifacts2))
158	print(artifacts1 - artifacts2)
159
160
161	if __name__ == "__main__":
162	asyncio.run(main())
163