initial commit

.github/workflows/python-app.yml

@@ -19,21 +19,19 @@ jobs:
 
     steps:
     - uses: actions/checkout@v4
-    - name: Set up Python 3.10
+    - name: Set up Python 3.9
       uses: actions/setup-python@v3
       with:
-        python-version: "3.10"
+        python-version: "3.9"
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install flake8 pytest
+        pip install black isort pytest
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
-    - name: Lint with flake8
+    - name: Check code compliance with black and isort
       run: |
-        # stop the build if there are Python syntax errors or undefined names
-        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
-        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+        black --check . 
+        isort --check-only .
     - name: Test with pytest
       run: |
         pytest

test_bum.py → field_notices_tests.py

@@ -1,11 +1,12 @@
 import json
+
 from selenium import webdriver
 from selenium.common import TimeoutException
-from selenium.webdriver.common.by import By
 from selenium.webdriver.chrome.service import Service
-from webdriver_manager.chrome import ChromeDriverManager
-from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.common.by import By
 from selenium.webdriver.support import expected_conditions as EC
+from selenium.webdriver.support.ui import WebDriverWait
+from webdriver_manager.chrome import ChromeDriverManager
 
 field_notices_link = "products-field-notices-list.html"
 eos_eos_notices_link = "eos-eol-notice-listing.html"
@@ -16,16 +17,20 @@
 driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))
 
 # Navigate to the Cisco support website
-driver.get('https://www.cisco.com/c/en/us/support/index.html')
+driver.get("https://www.cisco.com/c/en/us/support/index.html")
 
 # Optionally, add a wait time to ensure the page loads completely
 driver.maximize_window()
 driver.implicitly_wait(2)
 
 # Locate the section containing the product type links
-section = driver.find_element(By.XPATH, '//div[@data-config-metrics-title="Products by Category"]')
-links = section.find_elements(By.TAG_NAME, 'a')
-product_list = [{"product": i.text, "url": i.get_attribute("href")} for i in links]
+section = driver.find_element(
+    By.XPATH, '//div[@data-config-metrics-title="Products by Category"]'
+)
+links = section.find_elements(By.TAG_NAME, "a")
+product_list = [
+    {"product": i.text, "url": i.get_attribute("href")} for i in links
+]
 # Initialize a dictionary to hold the products by category
 products_by_category = {}
 
@@ -45,25 +50,22 @@
     # Find all product links in the "All Supported Products" section
     all_supported_products = driver.find_element(By.ID, "allSupportedProducts")
 
-    product_links = all_supported_products.find_elements(By.TAG_NAME, 'a')
+    product_links = all_supported_products.find_elements(By.TAG_NAME, "a")
 
     # Extract the product names and URLs
     for product_link in product_links:
         product_name = product_link.text
 
-        product_url = product_link.get_attribute('href')
+        product_url = product_link.get_attribute("href")
 
-        products.append({
-            "name": product_name,
-            "url": product_url
-        })
+        products.append({"name": product_name, "url": product_url})
     products_by_category[product["product"]] = products
 
 # Close the WebDriver
 driver.quit()
 
 # Save the products by category to a JSON file
-with open('products_by_category.json', 'w') as f:
+with open("products_by_category.json", "w") as f:
     json.dump(products_by_category, f, indent=4)
 
 print("JSON file has been created successfully.")

jobs/get_features.py

@@ -5,12 +5,14 @@
 import tarfile
 from pathlib import Path
 from typing import Any, Dict, List, Optional
+
 import aiofiles
 import httpx
 from pydantic import BaseModel
 
 logging.basicConfig(
-    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
 )
 
 # Suppress httpx logs by setting their logging level to WARNING
@@ -24,7 +26,9 @@ class Config:
     FETCH_FEATURES_ONLINE = True
     CONCURRENT_REQUESTS_LIMIT = 5
     REQUEST_DELAY = 1
-    FEATURES_DIR: Path = Path(os.path.join(os.getcwd(), "data", "product_features"))
+    FEATURES_DIR: Path = Path(
+        os.path.join(os.getcwd(), "data", "product_features")
+    )
     FEATURES_DIR.mkdir(exist_ok=True, parents=True)
     TYPES = [
         "Switches",
@@ -91,7 +95,10 @@ async def _fetch_platforms(
         return {}
 
     async def _fetch_releases(
-        self, client: httpx.AsyncClient, each_platform: Dict[str, Any], each_type: str
+        self,
+        client: httpx.AsyncClient,
+        each_platform: Dict[str, Any],
+        each_type: str,
     ) -> List[Dict[str, Any]]:
         """
         Fetch releases for the given platform.
@@ -102,7 +109,9 @@ async def _fetch_releases(
         """
         platform_id = each_platform.get("platform_id")
         self.logger.info(f"Fetching releases for {platform_id}")
-        request = RequestModel(platform_id=platform_id, mdf_product_type=each_type)
+        request = RequestModel(
+            platform_id=platform_id, mdf_product_type=each_type
+        )
         response = await client.post(
             self.config.REQUEST_2,
             headers=self.config.HEADERS,
@@ -157,9 +166,7 @@ async def _fetch_features(
                 feature["platform_id"] = each_release["platform_id"]
                 feature["release_id"] = each_release["release_id"]
 
-            file_name = (
-                f"{each_release['platform_id']}_{each_release['release_id']}.json"
-            )
+            file_name = f"{each_release['platform_id']}_{each_release['release_id']}.json"
             file_path = self.config.FEATURES_DIR / file_name
 
             async with aiofiles.open(file_path, "w") as file:
@@ -172,7 +179,9 @@ async def _fetch_features(
                 f"Failed to fetch features for platform {each_release['platform_id']} and release {each_release['release_id']}, status code: {response.status_code}"
             )
 
-    async def _read_file(self, filename: str) -> Dict[str, List[Dict[str, Any]]]:
+    async def _read_file(
+        self, filename: str
+    ) -> Dict[str, List[Dict[str, Any]]]:
         """
         Read data from a local JSON file.
         :param filename: The name of the file to read.
@@ -194,7 +203,9 @@ async def _fetch_all_features(
         :param tar: The tar file to add the features data.
         """
         async with httpx.AsyncClient(timeout=900) as client:
-            semaphore = asyncio.Semaphore(self.config.CONCURRENT_REQUESTS_LIMIT)
+            semaphore = asyncio.Semaphore(
+                self.config.CONCURRENT_REQUESTS_LIMIT
+            )
 
             async def fetch_features_with_semaphore(
                 each_release: Dict[str, Any], mdf_product_type: str
@@ -208,7 +219,9 @@ async def fetch_features_with_semaphore(
             for mdf_product_type, releases_list in releases.items():
                 for each_release in releases_list:
                     feature_tasks.append(
-                        fetch_features_with_semaphore(each_release, mdf_product_type)
+                        fetch_features_with_semaphore(
+                            each_release, mdf_product_type
+                        )
                     )
             await asyncio.gather(*feature_tasks)
             self.logger.info("Fetched all features data")
@@ -231,7 +244,9 @@ async def _fetch_platforms_data(self) -> Dict[str, Any]:
             return await self._fetch_online_platforms()
         return await self._read_file("platforms")
 
-    async def _fetch_releases_data(self, platforms: Dict[str, Any]) -> Dict[str, Any]:
+    async def _fetch_releases_data(
+        self, platforms: Dict[str, Any]
+    ) -> Dict[str, Any]:
         """
         Fetch or read releases data.
         :param platforms: A dictionary containing platforms data.
@@ -263,11 +278,15 @@ async def _fetch_online_platforms(self) -> Dict[str, Any]:
             platforms_results = await asyncio.gather(*platform_tasks)
             platforms = {
                 each_type: data
-                for each_type, data in zip(self.config.TYPES, platforms_results)
+                for each_type, data in zip(
+                    self.config.TYPES, platforms_results
+                )
             }
         return platforms
 
-    async def _fetch_online_releases(self, platforms: Dict[str, Any]) -> Dict[str, Any]:
+    async def _fetch_online_releases(
+        self, platforms: Dict[str, Any]
+    ) -> Dict[str, Any]:
         """
         Fetch releases data from the online API for the given platforms.
         :param platforms: A dictionary containing platforms data.
@@ -283,7 +302,9 @@ async def _fetch_online_releases(self, platforms: Dict[str, Any]) -> Dict[str, A
                 ]
                 releases_results = await asyncio.gather(*release_tasks)
                 releases[each_type] = [
-                    release for sublist in releases_results for release in sublist
+                    release
+                    for sublist in releases_results
+                    for release in sublist
                 ]
                 self.logger.info(
                     f"Retrieved {len(releases[each_type])} releases for {each_type}"

requirements.txt

@@ -3,4 +3,5 @@ httpx~=0.27.0
 pydantic~=2.7.4
 uvicorn~=0.30.1
 fastapi~=0.111.0
-starlette~=0.37.2
+starlette~=0.37.2
+selenium

routers/features.py

@@ -2,7 +2,7 @@
 import os
 from typing import Optional
 
-from fastapi import Depends, Query, HTTPException, APIRouter, Path
+from fastapi import APIRouter, Depends, HTTPException, Path, Query
 
 from models import Platform
 from utils import PaginationParams, extract_feature, paginate
@@ -11,7 +11,9 @@
 
 
 @features_router.get(
-    "/platforms", summary="Get platforms", description="Retrieve a list of platforms"
+    "/platforms",
+    summary="Get platforms",
+    description="Retrieve a list of platforms",
 )
 @paginate
 def features_platforms(
@@ -33,7 +35,9 @@ def features_platforms(
 
 
 @features_router.get(
-    "/releases", summary="Get releases", description="Retrieve a list of releases"
+    "/releases",
+    summary="Get releases",
+    description="Retrieve a list of releases",
 )
 @paginate
 def get_releases(
@@ -68,7 +72,9 @@ def get_features(
     file_name = f"{platform_id}_{release_id}.json"
 
     if not os.path.exists(tar_path):
-        raise HTTPException(status_code=404, detail="Feature archive not found.")
+        raise HTTPException(
+            status_code=404, detail="Feature archive not found."
+        )
 
     features = extract_feature(tar_path, file_name)
     return features

server.py

@@ -1,6 +1,7 @@
 import asyncio
 import threading
 from contextlib import asynccontextmanager
+
 import uvicorn
 from fastapi import FastAPI
 from starlette.responses import RedirectResponse

utils.py

@@ -2,9 +2,9 @@
 import json
 import tarfile
 from functools import lru_cache, wraps
-from typing import Callable, Union, Any, Dict, List
+from typing import Any, Callable, Dict, List, Union
 
-from fastapi import Query, Depends, HTTPException
+from fastapi import Depends, HTTPException, Query
 from fastapi.encoders import jsonable_encoder
 from fastapi.logger import logger
 from starlette.responses import JSONResponse
@@ -13,9 +13,14 @@
 class PaginationParams:
     def __init__(
         self,
-        page: int = Query(1, ge=1, description="The page number for pagination"),
+        page: int = Query(
+            1, ge=1, description="The page number for pagination"
+        ),
         limit: int = Query(
-            20, ge=1, le=1000, description="Limit the number of results per page"
+            20,
+            ge=1,
+            le=1000,
+            description="Limit the number of results per page",
         ),
     ):
         self.page = page
@@ -28,7 +33,9 @@ def offset(self):
 
 def paginate(func: Callable[..., Union[List[Dict[str, Any]], Dict[str, Any]]]):
     @wraps(func)
-    async def async_wrapper(*args, pagination: PaginationParams = Depends(), **kwargs):
+    async def async_wrapper(
+        *args, pagination: PaginationParams = Depends(), **kwargs
+    ):
         limit = pagination.limit
         offset = pagination.offset
 
@@ -41,9 +48,13 @@ async def async_wrapper(*args, pagination: PaginationParams = Depends(), **kwarg
             total_items = len(results)
             paginated_results = results[offset : offset + limit]
         else:
-            raise HTTPException(status_code=500, detail="Results should be a list.")
+            raise HTTPException(
+                status_code=500, detail="Results should be a list."
+            )
 
-        total_pages = (total_items + limit - 1) // limit  # Calculate total pages
+        total_pages = (
+            total_items + limit - 1
+        ) // limit  # Calculate total pages
         current_page = pagination.page
         has_more = offset + limit < total_items