chore(): Improves code features

* Upgrades python to 3.11.3
* Upgrade poetry configuration
* Upgrade project packages
* Introduces Upstream class
* Introduces an improved Registry class
* Introduces a common library for serialization
* Introduces Tags and Tag classes
* Introduces Images and Image classes
This commit is contained in:
Elia el Lazkani 2023-06-03 21:53:37 +02:00
parent c1f24cc67e
commit d8e7035369
9 changed files with 252 additions and 51 deletions

3
.gitignore vendored
View file

@ -1 +1,2 @@
/dist/
__pycache__/
dist/

16
poetry.lock generated
View file

@ -40,7 +40,6 @@ mypy-extensions = ">=0.4.3"
packaging = ">=22.0"
pathspec = ">=0.9.0"
platformdirs = ">=2"
tomli = {version = ">=1.1.0", markers = "python_version < \"3.11\""}
[package.extras]
colorama = ["colorama (>=0.4.3)"]
@ -265,17 +264,6 @@ urllib3 = ">=1.21.1,<3"
socks = ["PySocks (>=1.5.6,!=1.5.7)"]
use-chardet-on-py3 = ["chardet (>=3.0.2,<6)"]
[[package]]
name = "tomli"
version = "2.0.1"
description = "A lil' TOML parser"
optional = false
python-versions = ">=3.7"
files = [
{file = "tomli-2.0.1-py3-none-any.whl", hash = "sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc"},
{file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"},
]
[[package]]
name = "urllib3"
version = "2.0.2"
@ -295,5 +283,5 @@ zstd = ["zstandard (>=0.18.0)"]
[metadata]
lock-version = "2.0"
python-versions = "^3.10"
content-hash = "3ce15cfbe19b5ec59661c7d60325e6c80e5324cd14c14a195c8fa6778178a5c7"
python-versions = "^3.11"
content-hash = "7663a6c1d53181bd0becb2a6bb05afd3adad01d4875f04d0255a734363d1673b"

View file

@ -11,7 +11,7 @@ packages = [{include = "src"}]
rt = 'src.main:main'
[tool.poetry.dependencies]
python = "^3.10"
python = "^3.11"
requests = "^2.31.0"
[tool.poetry.group.dev.dependencies]

15
src/common.py Normal file
View file

@ -0,0 +1,15 @@
from datetime import datetime
class Serialize:
def boolean(string: str) -> bool:
if string == "true" or string == "True":
return True
return False
def integer(string: str) -> int:
return int(string)
def time(string: str) -> datetime:
return datetime.strptime(string, '%Y-%m-%dT%H:%M:%S.%f%z')

78
src/images.py Normal file
View file

@ -0,0 +1,78 @@
from typing import Self
from src.common import Serialize
class Images:
"""
Images class object.
"""
def __init__(self, images: list) -> None:
"""
Initialize an images object.
"""
self.images = []
self.deserialize(images=images)
def deserialize(self, images: list) -> Self:
"""
Method to deserialize images list.
"""
for image in images:
self.images.append(Image(image=image))
return self
def serialize(self) -> list:
"""
Method to serialize images object.
"""
images = []
for image in self.images:
images.append(image.serialize())
return images
class Image:
"""
Image class object.
"""
def __init__(self, image: dict) -> None:
"""
Initialize an image object.
"""
self.architecture = "arm"
self.features = ""
self.variant = "v7"
self.digest = "sha256:2627e55acb9ac183f1c94e6a44f869620d164bbb10d7c42b29df08513eb20c29"
self.os = "linux"
self.os_features = ""
self.os_version = None
self.size = 2911117
self.status = "active"
self.last_pulled = "2023-06-03T15:17:33.336554Z"
self.last_pushed = "2023-05-09T23:13:46.10789Z"
self.deserialize(image)
def deserialize(self, image: dict) -> Self:
"""
Method to deserialize image dictionary.
"""
self.architecture = image.get('architecture', None)
self.features = image.get('features', None)
self.variant = image.get('variant', None)
self.digest = image.get('digest', None)
self.os = image.get('os', None)
self.os_features = image.get('os_features', None)
self.os_version = image.get('os_version', None)
self.size = Serialize.integer(image.get('size', -1))
self.status = image.get('status', None)
self.last_pulled = Serialize.time(image.get('last_pulled', None))
self.last_pushed = Serialize.time(image.get('last_pushed', None))
return self
def serialize(self) -> dict:
"""
Method to serialize image object.
"""
return self.__dict__

View file

@ -1,16 +1,15 @@
#!/usr/bin/env python3
from src import registry
from src.registry import Registry
username = None
password = None
def main():
reg = registry.Registry(username=username, password=password)
tags = reg.get_tags_page("alpine")
print(tags)
reg = Registry(username=username, password=password)
registry = reg.populate_tags("traefik", namespace="rapidfort", url_params="page_size=25")
print(registry.tags.serialize())
if __name__ == "__main__":
main()

View file

@ -1,38 +1,27 @@
import json
from src import connection
from typing import Self
from src.upstream import Upstream
from src.tags import Tags
class Registry:
"""
Registry class.
class Registry(Upstream):
"""
Registry class
Keeps track of the image tags pulled from Upstream
"""
def __init__(
self,
url: str = "https://hub.docker.com",
username: str = None,
password: str = None,
) -> None:
self,
url: str = "https://hub.docker.com",
username: str = None,
password: str = None,
) -> None:
"""
Initializing the registry object.
Initialization method.
"""
self.url = url
self.req = connection.Requests()
if username and password:
self.req.get_token(url + "/v2/users/login", username, password)
self.tags = None
Upstream.__init__(self, url=url, username=username, password=password)
def get_tags_page(self, image, namespace: str = "library") -> dict:
"""
Method which returns the tags page in a dictionary.
"""
url = "{registry_url}/v2/namespaces/{namespace}/repositories/{repository}/tags".format(
registry_url=self.url, namespace=namespace, repository=image
)
try:
result = self.req.get(url, headers=self.req.get_headers())
except Exception as e:
print(e)
if not result.status_code == 200:
raise Exception("Could not get tags from server")
def populate_tags(self, image: str, namespace: str = None, url_params: str = None) -> Self:
tags = self.get_tags(image=image, namespace=namespace, url_params=url_params)
self.tags = Tags(tags)
return result.json()["results"]
return self

88
src/tags.py Normal file
View file

@ -0,0 +1,88 @@
from typing import Self
from src.images import Images
from src.common import Serialize
class Tags:
"""
Tags class object.
"""
def __init__(self, tags: list) -> None:
"""
Initialize a tag object.
"""
self.tags = []
self.deserialize(tags=tags)
def deserialize(self, tags: list) -> Self:
"""
Method to deserialize tags list.
"""
for tag in tags:
self.tags.append(Tag(tag=tag))
return self
def serialize(self) -> list:
"""
Method to serialize tags object.
"""
tags = []
for tag in self.tags:
tags.append(tag.serialize())
return tags
class Tag:
"""
Tag class object.
"""
def __init__(self, tag: dict) -> None:
"""
Initialize a tag object.
"""
self.creator = 7
self.id = 170608
self.images = None
self.last_updated = "2023-05-09T23:14:23.530698Z"
self.last_updater = 1156886
self.last_updater_username = "doijanky"
self.name = "latest"
self.repository = 160398
self.full_size = 3397490
self.v2 = True
self.tag_status = "active"
self.tag_last_pulled = "2023-06-03T16:01:48.740291Z"
self.tag_last_pushed = "2023-05-09T23:14:23.530698Z"
self.media_type = "application/vnd.docker.d…n.manifest.list.v2+json"
self.content_type = "image"
self.digest = "sha256:02bb6f428431fbc28…508869a33cb1af4444c9b11"
self.deserialize(tag)
def deserialize(self, tag: dict) -> Self:
"""
Method to parse tag dictionary.
"""
self.creator = Serialize.integer(tag.get('creator', -1))
self.id = Serialize.integer(tag.get('id', -1))
self.images = Images(tag.get("images", None))
self.last_updated = Serialize.time(tag.get("last_updated", None))
self.last_updater = Serialize.integer(tag.get("last_updater", None))
self.last_updater_username = tag.get("last_updater_username", None)
self.name = tag.get("name", None)
self.repository = Serialize.integer(tag.get("repository", -1))
self.full_size = Serialize.integer(tag.get("full_size", -1))
self.v2 = Serialize.boolean(tag.get("v2", None))
self.tag_status = tag.get("tag_status", None)
self.tag_last_pulled = Serialize.time(tag.get("tag_last_pulled", None))
self.tag_last_pushed = Serialize.time(tag.get("tag_last_pushed", None))
self.media_type = tag.get("media_type", None)
self.content_type = tag.get("content_type", None)
self.digest = tag.get("digest", None)
return self
def serialize(self) -> dict:
"""
Method to serialize tag object.
"""
return self.__dict__

43
src/upstream.py Normal file
View file

@ -0,0 +1,43 @@
import json
from src.connection import Requests
class Upstream:
"""
Registry class.
"""
def __init__(
self,
url: str = "https://hub.docker.com",
username: str = None,
password: str = None,
) -> None:
"""
Initialize the upstream registry object.
"""
self.url = url
self.req = Requests()
if username and password:
self.req.get_token(url + "/v2/users/login", username, password)
def get_tags(self, image, namespace: str = "library", url_params: str = None) -> dict:
"""
Method which returns the tags page in a dictionary.
"""
if url_params:
url = "{registry_url}/v2/namespaces/{namespace}/repositories/{repository}/tags?{url_params}".format(
registry_url=self.url, namespace=namespace, repository=image, url_params=url_params
)
else:
url = "{registry_url}/v2/namespaces/{namespace}/repositories/{repository}/tags".format(
registry_url=self.url, namespace=namespace, repository=image
)
try:
result = self.req.get(url, headers=self.req.get_headers())
except Exception as e:
print(e)
if not result.status_code == 200:
raise Exception("Could not get tags from server")
return result.json()["results"]