Unified log levels
- Logger doesn't start with log level DEBUG by default, instead reads a environment variable - Secret handler raises exceptions instead of using the module os to exit - Added extensive debug logging - Notifications: Added timestamp and category
This commit is contained in:
1 parent
3a722eca04
commit
f2ffe394ad
9 files changed
+249
-99
No files matched your search
+52
-14
@@ -1,29 +1,67 @@
|
||||
import requests
|
||||
from secret_handler import return_credentials
|
||||
from logger_handler import setup_logger
|
||||
|
||||
dockerhub_token = return_credentials("/etc/secrets/dockerhub_token")
|
||||
dockerhub_username = return_credentials("/etc/secrets/dockerhub_username")
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
def login_and_get_token()->str:
|
||||
def login_and_get_token() -> str:
|
||||
"""
|
||||
You have to first login with credentials to get a token for subsequent api requests.
|
||||
[DockerHub] Login to Docker Hub and retrieve a JWT token.
|
||||
|
||||
Returns:
|
||||
A string JWT token for authenticated requests.
|
||||
Returns None and logs error if login fails.
|
||||
"""
|
||||
login_url = "https://hub.docker.com/v2/users/login/"
|
||||
response = requests.post(login_url,
|
||||
json={"username": dockerhub_username, "password": dockerhub_token})
|
||||
if response.status_code == 200:
|
||||
token = response.json()["token"]
|
||||
return token
|
||||
else:
|
||||
print(f"Login failed: {response.status_code} - {response.text}")
|
||||
response = requests.post(
|
||||
login_url,
|
||||
json={"username": dockerhub_username, "password": dockerhub_token}
|
||||
)
|
||||
|
||||
def find_package_version_with_tag(repo, tag):
|
||||
if response.status_code == 200:
|
||||
token = response.json().get("token")
|
||||
if token:
|
||||
logger.debug("[DockerHub] Login successful, token obtained")
|
||||
return token
|
||||
else:
|
||||
logger.error("[DockerHub] Login response did not contain token")
|
||||
return None
|
||||
else:
|
||||
logger.error(f"[DockerHub] Login failed: {response.status_code} - {response.text}")
|
||||
return None
|
||||
|
||||
|
||||
def find_package_version_with_tag(repo: str, tag: str) -> str:
|
||||
"""
|
||||
[DockerHub] Fetch the tag ID for a repository tag.
|
||||
|
||||
Authenticates with Docker Hub, queries the repository for the tag,
|
||||
returns the tag ID if found, else None.
|
||||
"""
|
||||
token = login_and_get_token()
|
||||
if not token:
|
||||
logger.error(f"[DockerHub] Cannot fetch tag '{tag}' for repo '{repo}' without token")
|
||||
return None
|
||||
|
||||
headers = {"Authorization": f"JWT {token}"}
|
||||
tags_url = f"https://hub.docker.com/v2/repositories/{repo}/tags/{tag}?page_size=1"
|
||||
tags_response = requests.get(tags_url, headers=headers)
|
||||
id = tags_response.json()["id"]
|
||||
return id
|
||||
|
||||
|
||||
response = requests.get(tags_url, headers=headers)
|
||||
if response.status_code != 200:
|
||||
logger.error(f"[DockerHub] Failed to fetch tag '{tag}' for repo '{repo}': {response.status_code} - {response.text}")
|
||||
return None
|
||||
|
||||
data = response.json()
|
||||
tag_id = data.get("id")
|
||||
if tag_id:
|
||||
logger.info(f"[DockerHub] Found tag '{tag}' for repo '{repo}', ID: {tag_id}")
|
||||
return tag_id
|
||||
else:
|
||||
logger.debug(f"[DockerHub] Tag '{tag}' not found for repo '{repo}'")
|
||||
return None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(find_package_version_with_tag("pihole/pihole", "latest"))
|
||||
Reference in new issue
Block a user