Arranging imports

Dockerfile

@@ -1,9 +0,0 @@
-FROM python:latest
-
-WORKDIR /usr/local/bin
-COPY requirements.txt requirements.txt
-RUN pip install -r requirements.txt
-
-COPY src/start.py .
-
-CMD ["src/start.py"]

config/inventory.yml

@@ -1,8 +0,0 @@
-- 1:
-    hostname: '172.16.57.2'
-    username: 'admin'
-    password: 'Netapp12'
-- 2:
-    hostname: '172.16.56.2'
-    username: 'admin'
-    password: 'Netapp12'

requirements.txt

@@ -1,7 +1,4 @@
 fastapi[standard]>=0.116.2
 httpx>=0.28.1
 redis>=6.4.0
-python-dotenv>=1.1.1
+python-dotenv>=1.1.1
-pydantic
-redis[hiredis]
-dotenv

src/aggregate/aggregate_router.py

@@ -1,16 +1,23 @@
 # contains the router for the aggregates endpoint
-from fastapi import APIRouter, Query, Request
+from enum import Enum
 from typing import List
-from .aggregate_schema import AggregateSchema, MetricEnum
+
+from fastapi import APIRouter, Query
+
+from .aggregate_schema import AggregateSchema
 from .aggregate_service import get_aggregates
 
 
+class MetricEnum(str, Enum):
+    relative = "relative"
+    absolute = "absolute"
+
+
 router = APIRouter(tags=["aggregates"])
 
 
 @router.get("/aggregates", response_model=List[AggregateSchema])
 async def aggregates_endpoint(
-    request: Request,
     metric: MetricEnum = Query(MetricEnum.relative, description="Metric type"),
 ):
-    return await get_aggregates(request, metric)
+    return await get_aggregates(metric)

src/aggregate/aggregate_schema.py

@@ -1,15 +1,8 @@
 # contains the schema definitions for aggregates
 from pydantic import BaseModel
-from enum import Enum
 
 
 class AggregateSchema(BaseModel):
     aggregate: str
     node: str
-    available: int
+    available: str
-    available_str: str
-
-
-class MetricEnum(str, Enum):
-    relative = "relative"
-    absolute = "absolute"

src/aggregate/aggregate_service.py

@@ -1,38 +1,25 @@
 # contains the business logic for aggregates
-
 from typing import List
 
-from fastapi import Request
+from .aggregate_schema import AggregateSchema
-from .aggregate_schema import AggregateSchema, MetricEnum
-from logging import getLogger
-from ..utils import round_bytes, get_data_from_ontap
-
-logger = getLogger("uvicorn")
-logger.setLevel("DEBUG")
 
 
-async def get_aggregates(request: Request, metric: str = "relative") -> List[AggregateSchema]:
+async def get_aggregates(metric: str = "relative") -> List[AggregateSchema]:
     # Dummy data for demonstration
     # You can use the metric parameter to filter or modify results as needed
     # For now, just return the same data and show metric usage
-    logger.debug(f"Metric used: {metric}")
+    print(f"Metric used: {metric}")
-    client = request.app.requests_client
-    __aggregates = await get_data_from_ontap(client, logger, "172.16.57.2", "admin", "Netapp12", "storage/aggregates", "fields=name,uuid,space,node,home_node")
-    logger.debug(__aggregates)
-    __aggregates = __aggregates.get("records")
-    if metric == MetricEnum.relative:
-        __aggregates = sorted(__aggregates, key=lambda r: r["space"]["block_storage"].get("used_percent"), reverse=True)
-    elif metric == MetricEnum.absolute:
-        __aggregates = sorted(__aggregates, key=lambda r: r["space"]["block_storage"].get("available"), reverse=False)
 
     aggregates: list = [
         AggregateSchema(
-            aggregate=a["name"],
+            aggregate="Aggregate A", node="cluster01-01", available="100.0TB"
-            node=a["node"]["name"],
+        ),
-            available=a["space"]["block_storage"]["available"],
+        AggregateSchema(
-            available_str=round_bytes(a["space"]["block_storage"]["available"]),
+            aggregate="Aggregate B", node="cluster01-01", available="200.5GB"
-        )
+        ),
-        for a in __aggregates
+        AggregateSchema(
+            aggregate="Aggregate C", node="cluster01-02", available="300.75MB"
+        ),
     ]
 
     return aggregates

src/database.py

@@ -1,40 +0,0 @@
-import json
-import logging
-from redis import Redis, ConnectionError
-from typing import List
-from pydantic import TypeAdapter
-from schema import ConfigSchema
-
-
-def setup_db_conn(redishost, redisport: str):
-    ''' Setup Redis connection and return it open'''
-    log = logging.getLogger('uvicorn')
-    try:
-        redisclient = Redis(host=redishost, port=redisport, decode_responses=True)
-        if redisclient.ping():
-            log.info(f"Connected to Redis DB {redishost} on port {redisport}")
-        else:
-            log.error(f"Cannot connect to Redis DB {redishost} on port {redisport}")
-            exit(1)
-        return redisclient
-    except ConnectionError as e:
-        print(f"FATAL: Redis DB {redishost} is unreachable on port {redisport}. Err: {e}")
-        return None
-    except Exception as e:
-        print(f"FATAL: {e}")
-        return None
-
-def get_inventory_from_redis(redisclient: Redis):
-    ''' Read inventory from Redis '''
-    cluster_inv = redisclient.hgetall('cluster_inventory')
-    if 'inventory' in cluster_inv:
-        return json.loads(cluster_inv['inventory'])
-    return {}
-
-def get_config_from_db(redisclient: Redis) -> ConfigSchema:
-    ''' Load inventory to global vars'''
-    GLOBAL_INVENTORY = get_inventory_from_redis(redisclient)
-
-    GLOBAL_INVENTORY_VALID = TypeAdapter(List[ConfigSchema]).validate_python(GLOBAL_INVENTORY)
-
-    return GLOBAL_INVENTORY_VALID

src/example/schema.py

@@ -5,11 +5,3 @@ from pydantic import BaseModel
 class ExampleSchema(BaseModel):
     example_field: str
     another_field: int
-
-class ClusterCreds(BaseModel):
-    """A structure to hold basic auth cluster credentials for a cluster"""
-    username:      str
-    password:      str
-    hostname:      str  = None
-    cert_filepath: Path = None
-    key_filepath:  Path = None

src/initialize.py

@@ -1,44 +0,0 @@
-import os
-import json
-import logging
-import yaml
-
-from pathlib import Path
-from dotenv import load_dotenv
-from database import setup_db_conn
-from schema import ConfigSchema
-from typing import List
-from pydantic import TypeAdapter
-
-def initialize_config():
-    load_dotenv()
-    log = logging.getLogger('uvicorn')
-    ENV_INVENTORYPATH = os.getenv('cluster_inventory_path')
-    ENV_REDISHOST = os.getenv('redis_host')
-    ENV_REDISPORT = os.getenv('redis_port')
-
-    log.info(f"Found Cluster Inventory file at: {ENV_INVENTORYPATH}")
-    if not ENV_INVENTORYPATH or not Path(ENV_INVENTORYPATH).is_file():
-        print(f"FATAL: Inventory file {ENV_INVENTORYPATH} is missing or not a file.")
-        return False
-    try:
-        with open(ENV_INVENTORYPATH, 'r') as f:
-            inv = yaml.safe_load(f)
-            inventory = json.dumps(inv) 
-    except Exception as e:
-        print(f"FATAL: Cannot read inventory file {ENV_INVENTORYPATH}. Err: {e}")
-        return False
-
-    print(f'[INFO] Importing configuration to DB...')
-    try:
-        GLOBAL_INVENTORY_VALID = TypeAdapter(List[ConfigSchema]).validate_python(inv)
-        redis_conn = setup_db_conn(ENV_REDISHOST, ENV_REDISPORT)
-        redis_conn.hset('cluster_inventory', mapping={'inventory': inventory})
-        redis_conn.close()
-
-        log.info("Configuration has been loaded.") 
-        return True
-
-    except Exception as e: 
-        print(f"FATAL: Redis DB error: {e}")
-        return False

src/main.py

@@ -1,48 +1,27 @@
-import os
 import logging
 
 from fastapi import FastAPI
 
 from src.aggregate import aggregate_router
 from src.config import config_router
-
+from src.service import load_config
-from contextlib import asynccontextmanager
-
-from database import setup_db_conn, get_config_from_db
-from src.initialize import initialize_config
-from utils import setup_logging
 
 logger = logging.getLogger("uvicorn")
+
 logger.info("Starting application")
+config = load_config()
 
 app = FastAPI()
 app.include_router(aggregate_router)
 app.include_router(config_router)
 
 
-@asynccontextmanager
+@app.get("/")
-async def lifespan(app: FastAPI):
+async def main():
-    """make loading it async"""
+    return {"Hello": "World"}
-    log = logging.getLogger("uvicorn")
-    cfg_init_result = initialize_config()
-
-    shared_redis_conn = setup_db_conn(os.getenv("redis_host"), os.getenv("redis_port"))
-    if not shared_redis_conn:
-        log.error("Cannot connect to Redis DB. Exiting...")
-        exit(1)
-
-    inv_check = get_config_from_db(shared_redis_conn)
-    log.info(f"[DEBUG] Data validity healthcheck (DEVELOPER MODE): {inv_check}")
-    if not cfg_init_result:
-        log.error("Configuration initialization failed. Exiting...")
-        # exit(1)
-
-    yield
-    log.info("Shutting down FastAPI app...")
 
 
-setup_logging()
+@app.get("/config")
-log = logging.getLogger("uvicorn")
+async def get_config():
-
+    """Endpoint to get the current configuration."""
-log.info("Starting FastAPI app...")
+    return config.model_dump()
-app = FastAPI(lifespan=lifespan)

src/utils.py

@@ -1,33 +0,0 @@
-import logging
-import httpx
-
-def round_bytes(size_in_bytes: int) -> str:
-    # Helper function to convert bytes to a human-readable format
-    for unit in ["B", "KiB", "MiB", "GiB", "TiB", "PiB"]:
-        if size_in_bytes < 1024:
-            return f"{size_in_bytes:.2f}{unit}"
-        size_in_bytes /= 1024
-    return f"{size_in_bytes:.2f}EB"
-
-
-async def get_data_from_ontap(client, logger, hostname: str, username: str, password: str, endpoint: str, query_string: str = ""):
-    url = f"https://{hostname}/api/{endpoint}"
-    if query_string:
-        url += f"?{query_string}"
-    async with client as _client:
-        try:
-            logger.debug(f"Fetching data from ONTAP: {url}")
-            response = await _client.get(url, auth=(username, password))
-            response.raise_for_status()
-            return response.json()
-        except httpx.HTTPError as e:
-            logger.error(f"HTTP error occurred: {e}")
-            return None
-
-def setup_logging() -> None:
-    """Configure logging for the application"""
-    logging.basicConfig(
-        level=logging.DEBUG,
-        format="[%(asctime)s] [%(levelname)5s] %(message)s"
-    )
-    print(f"Logger is initialized.")