Merge pull request #1 from maxhoheiser/feature/dockerHubImageStats

Author: maxhoheiser

src/__init__.py

@@ -0,0 +1,48 @@
+import logging
+from typing import Tuple
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+class DockerHubService:
+    """
+    Handles read and write DockerHub API operations
+    """
+
+    def __init__(self, user_name: str, image_name: str) -> None:
+        self.user_name = user_name
+        self.image_name = image_name
+
+    def get_dockerhub_image_stats(self) -> Tuple[int, int]:
+        api_url = f"https://hub.docker.com/v2/repositories/{self.user_name}/{self.image_name}/"
+
+        try:
+            response = requests.get(api_url)
+        except Exception as e:
+            logger.error(f"Error: Unable to fetch data. Exception: {e}")
+            raise
+
+        if response.status_code == 200:
+            image_data = response.json()
+            pull_count = image_data["pull_count"]
+            star_count = image_data["star_count"]
+            return pull_count, star_count
+        else:
+            logger.error(
+                f"Error: Unable to fetch data. Status Code: {response.status_code}"
+            )
+            raise
+
+
+if __name__ == "__main__":
+    user_name = "localstack"
+    image_name = "localstack"
+
+    LocalStackDockerHubService = DockerHubService(user_name, image_name)
+    pull_count, star_count = LocalStackDockerHubService.get_dockerhub_image_stats()
+
+    if pull_count is not None and star_count is not None:
+        print(f"Pull Count: {pull_count}")
+        print(f"Star Count: {star_count}")

src/dockerHubStats/__init__.py

@@ -0,0 +1,39 @@
+import unittest
+from typing import Any, Dict
+from unittest.mock import MagicMock, patch
+
+from src.dockerHubStats.dockerHubStatsService import DockerHubService
+
+USER_NAME: str = "localstack"
+IMAGE_NAME: str = "localstack"
+
+
+class TestDockerHubService(unittest.TestCase):
+    def setUp(self) -> None:
+        self.user_name: str = USER_NAME
+        self.image_name: str = IMAGE_NAME
+        self.service: DockerHubService = DockerHubService(
+            self.user_name, self.image_name
+        )
+
+    @patch("src.dockerHubStats.dockerHubStatsService.requests.get")
+    def test_get_dockerhub_image_stats_success(self, mock_get: MagicMock) -> None:
+        mock_response: Dict[str, Any] = {"pull_count": 100, "star_count": 50}
+        mock_get.return_value.status_code = 200
+        mock_get.return_value.json.return_value = mock_response
+
+        pull_count, star_count = self.service.get_dockerhub_image_stats()
+
+        self.assertEqual(pull_count, 100)
+        self.assertEqual(star_count, 50)
+
+    @patch("src.dockerHubStats.dockerHubStatsService.requests.get")
+    def test_get_dockerhub_image_stats_error(self, mock_get: MagicMock) -> None:
+        mock_get.return_value.status_code = 404
+
+        with self.assertRaises(Exception):
+            self.service.get_dockerhub_image_stats()
+
+
+if __name__ == "__main__":
+    unittest.main()