harbor_paginate.py

harbor_paginate.py · 5.5 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):
    iterator = iter(lst)
    return list(iter(lambda: list(islice(iterator, 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,
        )
        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}")
            # режем на порции
            for portion in chunks(tags, page_size):
                # если достигли нужной страницы или получили пустышку от Harbor
                if rim_page_number == page_number:
                    # возвращаем страницы
                    return artifacts_page_buffer.get_filled()
                # буфер страницы заполняется
                artifacts_page_buffer.load(portion)
                # если в накопителе тэгов не исчерпались элементы
                if artifacts_page_buffer.filled_size() == page_size:
                    # увеличиваем счетчик страниц RIM
                    rim_page_number += 1
        # исчерпали все артефакты
        if len(artifacts) == 0:
            return artifacts_page_buffer.get_filled()
        # увеличиваем счетчик страниц Harbor
        harbor_page_number += 1

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"
    artifacts = await get_artifacts(client, project_name, repository_name, 1, 21)
    print(artifacts)
    print(len(artifacts))

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	iterator = iter(lst)
57	return list(iter(lambda: list(islice(iterator, 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	for artifact in artifacts:
84	# массив тэгов
85	tags = []
86	# если есть тэги
87	if artifact.tags:
88	# перебираем все ссылки на тэги
89	for tag_reference in artifact.tags:
90	# формируем имя
91	artifact_name = (
92	f"{project_name}/{repository_name}:{tag_reference.name}"
93	)
94	# засовываем в список
95	tags.append(artifact_name)
96	print(f"Tags {tags}")
97	print(f"Buffer: {artifacts_page_buffer.get_filled()}")
98	print(f"RIM page number: {rim_page_number}")
99	# режем на порции
100	for portion in chunks(tags, page_size):
101	# если достигли нужной страницы или получили пустышку от Harbor
102	if rim_page_number == page_number:
103	# возвращаем страницы
104	return artifacts_page_buffer.get_filled()
105	# буфер страницы заполняется
106	artifacts_page_buffer.load(portion)
107	# если в накопителе тэгов не исчерпались элементы
108	if artifacts_page_buffer.filled_size() == page_size:
109	# увеличиваем счетчик страниц RIM
110	rim_page_number += 1
111	# исчерпали все артефакты
112	if len(artifacts) == 0:
113	return artifacts_page_buffer.get_filled()
114	# увеличиваем счетчик страниц Harbor
115	harbor_page_number += 1
116
117
118	async def main():
119	registry_url = "-"
120	registry_login = "-"
121	registry_password = "-"
122	client = HarborAsyncClient(
123	url=registry_url + "/api/v2.0",
124	username=registry_login,
125	secret=registry_password,
126	# ограничение по количеству количества записей о соединениях хранимых клиентом
127	max_logs=0,
128	verify=False, # TODO: remove them ?
129	)
130	rbuf = AccumulatingBuffer(10)
131	rbuf.load([])
132	print(rbuf.get_filled())
133	rbuf.load([1, 2, 3, 4, 5, 6, 7, 8, 9])
134	print(rbuf.get_filled())
135	rbuf.load([1, 2, 3, 4])
136	print(rbuf.get_filled())
137	# tests
138	project_name = "sbis-hlha"
139	repository_name = "charts/sbis-docker-dispatcher"
140	artifacts = await get_artifacts(client, project_name, repository_name, 1, 21)
141	print(artifacts)
142	print(len(artifacts))
143
144
145	if __name__ == "__main__":
146	asyncio.run(main())