Merge pull request 'main' (#1) from CIzz22/rbd-app:main into main

Reviewed-on: DigitalTwin/rbd-app#1

.env

@@ -16,5 +16,7 @@ COLLECTOR_CREDENTIAL_PASSWORD=postgres
 COLLECTOR_NAME=digital_aeros_fixed
 
 
+AEROS_LICENSE_ID=20260218-Jre5VZieQfWXTq0G8ClpVSGszMf4UEUMLS5ENpWRVcoVSrNJckVZzXE
+AEROS_LICENSE_SECRET=GmLIxf9fr8Ap5m1IYzkk4RPBFcm7UBvcd0eRdRQ03oRdxLHQA0d9oyhUk2ZlM3LVdRh1mkgYy5254bmCjFyWWc0oPFwNWYzNwDwnv50qy6SLRdaFnI0yZcfLbWQ7qCSj
 WINDOWS_AEROS_BASE_URL=http://192.168.1.102:8800
-TEMPORAL_URL=http://192.168.1.86:7233
+TEMPORAL_URL=http://192.168.1.86:7233

poetry.lock

@@ -1076,6 +1076,27 @@ MarkupSafe = ">=2.0"
 [package.extras]
 i18n = ["Babel (>=2.7)"]
 
+[[package]]
+name = "licaeros"
+version = "0.1.0"
+description = ""
+optional = false
+python-versions = "*"
+groups = ["main"]
+files = [
+    {file = "licaeros-0.1.0-cp310-cp310-linux_x86_64.whl", hash = "sha256:7ab820555c10edae6d8f391e71f0188463283ddbf60840a3baadb926bb78a6a9"},
+    {file = "licaeros-0.1.0-cp311-cp311-linux_x86_64.whl", hash = "sha256:70b6b753c96e6fa4912e0e9eddf8088ddc3c4b6947ca16dc4b19047d2ee4aedf"},
+    {file = "licaeros-0.1.0-cp312-cp312-linux_x86_64.whl", hash = "sha256:d89fe52a78637f2a72abf64f6a74445aad3899303cce7409f2d12fd277a0db00"},
+]
+
+[package.dependencies]
+requests = "*"
+
+[package.source]
+type = "legacy"
+url = "https://git.reliabilityindonesia.com/api/packages/DigitalTwin/pypi/simple"
+reference = "licaeros-repo"
+
 [[package]]
 name = "limits"
 version = "3.13.0"
@@ -2823,4 +2844,4 @@ propcache = ">=0.2.1"
 [metadata]
 lock-version = "2.1"
 python-versions = "^3.11"
-content-hash = "9d61d2415a02d2e9a0be0850394cd0d8dbf178f7a419e726cc9b3b1c37d8d4d1"
+content-hash = "f75002a661bdae021c2906c1b44ba85cfe1835d37378a717ee9561e376603771"

pyproject.toml

@@ -32,8 +32,14 @@ aiohttp = "^3.12.14"
 ijson = "^3.4.0"
 redis = "^7.1.0"
 clamd = "^1.0.2"
+licaeros = "^0.1.0"
 
 
+[[tool.poetry.source]]
+name = "licaeros-repo"
+url = "https://git.reliabilityindonesia.com/api/packages/DigitalTwin/pypi/simple/"
+priority = "supplemental"
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"

run.py

@@ -3,4 +3,4 @@ import uvicorn
 from src.config import HOST, PORT
 
 if __name__ == "__main__":
-    uvicorn.run("src.main:app", host=HOST, port=PORT, reload=True)
+    uvicorn.run("src.main:app", host=HOST, port=PORT, reload=True,)

src/aeros_equipment/router.py

@@ -14,7 +14,7 @@ from .service import save_default_equipment, get_all
 
 # from .schema import (OverhaulScheduleCreate, OverhaulSchedulePagination, OverhaulScheduleUpdate)
 
-router = APIRouter()
+router = APIRouter()    
 
 
 @router.get("", response_model=StandardResponse[EquipmentPagination])

src/aeros_equipment/service.py

@@ -7,7 +7,8 @@ from sqlalchemy import Delete, Select, func, desc, and_, text
 from sqlalchemy.orm import selectinload
 
 from src.auth.service import CurrentUser
-from src.config import AEROS_BASE_URL, DEFAULT_PROJECT_NAME, RELIABILITY_SERVICE_API
+from src.config import DEFAULT_PROJECT_NAME, RELIABILITY_SERVICE_API
+from src.aeros_utils import aeros_post
 from src.database.core import CollectorDbSession, DbSession
 from src.database.service import search_filter_sort_paginate
 from .model import AerosEquipment, AerosEquipmentDetail, MasterEquipment, AerosEquipmentGroup, ReliabilityPredictNonRepairable
@@ -21,7 +22,7 @@ import json
 import pandas as pd
 from src.aeros_simulation.service import get_aeros_schematic_by_name
 
-client = httpx.AsyncClient(timeout=300.0)
+# Aeros session is managed in src.aeros_utils
 log = logging.getLogger()
 
 async def get_project(*, db_session: DbSession):
@@ -47,11 +48,13 @@ async def get_all(*, common):
     updateNodeReq = {"projectName": project.project_name , "equipmentNames": reg_nodes}
 
     try:
-        response = await client.post(
+        response = await aeros_post(
-            f"{AEROS_BASE_URL}/api/UpdateDisParams/GetUpdatedNodeDistributions",
+            "/api/UpdateDisParams/GetUpdatedNodeDistributions",
             json=updateNodeReq,
             headers={"Content-Type": "application/json"},
         )
+
+        
         response.raise_for_status()
 
         res = response.json()
@@ -96,8 +99,8 @@ async def get_by_id(*, db_session: DbSession, id: UUID):
     }
 
     try:
-        response = await client.post(
+        response = await aeros_post(
-            f"{AEROS_BASE_URL}/api/UpdateDisParams/GetUpdatedNodeDistributions",
+            "/api/UpdateDisParams/GetUpdatedNodeDistributions",
             json=aerosNodeReq,
             headers={"Content-Type": "application/json"},
         )
@@ -124,8 +127,8 @@ async def get_aeros_equipment_by_location_tag(*, location_tag: Union[str, List[s
     }
 
     try:
-        response = await client.post(
+        response = await aeros_post(
-            f"{AEROS_BASE_URL}/api/UpdateDisParams/GetUpdatedNodeDistributions",
+            "/api/UpdateDisParams/GetUpdatedNodeDistributions",
             json=aerosNodeReq,
             headers={"Content-Type": "application/json"},
         )
@@ -157,8 +160,8 @@ async def update_node(
 
 
     try:
-        response = await client.post(
+        response = await aeros_post(
-            f"{AEROS_BASE_URL}/api/UpdateDisParams/UpdateEquipmentDistributions",
+            "/api/UpdateDisParams/UpdateEquipmentDistributions",
             json=updateNodeReq,
             headers={"Content-Type": "application/json"},
         )
@@ -188,8 +191,8 @@ async def save_default_equipment(*, db_session: DbSession, project_name: str):
     await db_session.execute(query)
 
     try:
-        response = await client.post(
+        response = await aeros_post(
-            f"{AEROS_BASE_URL}/api/UpdateDisParams/GetUpdatedNodeDistributions",
+            "/api/UpdateDisParams/GetUpdatedNodeDistributions",
             json=updateNodeReq,
             headers={"Content-Type": "application/json"},
         )

src/aeros_project/service.py

@@ -9,7 +9,8 @@ from sqlalchemy.orm import selectinload
 from src.aeros_equipment.service import save_default_equipment
 from src.aeros_simulation.service import save_default_simulation_node
 from src.auth.service import CurrentUser
-from src.config import WINDOWS_AEROS_BASE_URL, AEROS_BASE_URL, CLAMAV_HOST, CLAMAV_PORT
+from src.config import WINDOWS_AEROS_BASE_URL, CLAMAV_HOST, CLAMAV_PORT
+from src.aeros_utils import aeros_post
 from src.database.core import DbSession
 from src.database.service import search_filter_sort_paginate
 from src.utils import sanitize_filename
@@ -21,6 +22,8 @@ from .schema import AerosProjectInput, OverhaulScheduleCreate, OverhaulScheduleU
 import asyncio
 ALLOWED_EXTENSIONS = {".aro"}
 MAX_FILE_SIZE = 100 * 1024 * 1024  # 100MB
+# client = httpx.AsyncClient(timeout=300.0) # Managed in src.aeros_utils
+# We still use a local client for WINDOWS_AEROS_BASE_URL if it's not managed by aeros_utils
 client = httpx.AsyncClient(timeout=300.0)
 
 
@@ -38,6 +41,14 @@ async def import_aro_project(*, db_session: DbSession, aeros_project_in: AerosPr
             status_code=400,
             detail=f"Invalid filename: {str(e)}"
         )
+    
+    # Check if mime type is application/octet-stream 
+    if file.content_type != "application/octet-stream":
+        raise HTTPException(
+            status_code=400,
+            detail="Invalid file type. Allowed: application/octet-stream"
+        )
+
 
     # Get filename
     filename_without_ext = os.path.splitext(clean_filename)[0]
@@ -158,9 +169,9 @@ async def import_aro_project(*, db_session: DbSession, aeros_project_in: AerosPr
     # Update path to AEROS APP
     # Example BODy "C/dsad/dsad.aro"
     try:
-        response = await client.post(
+        response = await aeros_post(
-            f"{AEROS_BASE_URL}/api/Project/ImportAROFile",
+            "/api/Project/ImportAROFile",
-            content=f'"{aro_path}"',
+            data=f'"{aro_path}"',
             headers={"Content-Type": "application/json"},
         )
         response.raise_for_status()
@@ -224,9 +235,9 @@ async def reset_project(*, db_session: DbSession):
     project = await fetch_aro_record(db_session=db_session)
 
     try:
-        response = await client.post(
+        response = await aeros_post(
-            f"{AEROS_BASE_URL}/api/Project/ImportAROFile",
+            "/api/Project/ImportAROFile",
-            content=f'"{project.aro_file_path}"',
+            data=f'"{project.aro_file_path}"',
             headers={"Content-Type": "application/json"},
         )
         response.raise_for_status()

src/aeros_simulation/service.py

@@ -8,11 +8,11 @@ import logging
 import httpx
 from fastapi import HTTPException, status
 import ijson
-import requests
 from sqlalchemy import delete, desc, select, update, and_
 from sqlalchemy.orm import selectinload
 
-from src.config import AEROS_BASE_URL, AEROS_BASE_URL_OLD, DEFAULT_PROJECT_NAME
+from src.config import DEFAULT_PROJECT_NAME
+from src.aeros_utils import aeros_post
 from src.database.core import DbSession
 from src.database.service import CommonParameters, search_filter_sort_paginate
 from src.utils import save_to_pastebin
@@ -32,7 +32,7 @@ from src.aeros_equipment.schema import EquipmentWithCustomParameters
 from .schema import SimulationInput, SimulationPlotResult, SimulationRankingParameters
 from .utils import calculate_eaf, stream_large_array
 
-client = httpx.AsyncClient(timeout=300.0)
+# client = httpx.AsyncClient(timeout=300.0) # Managed in src.aeros_utils
 active_simulations = {}
 
 
@@ -399,11 +399,11 @@ async def execute_simulation(*, db_session: DbSession, simulation_id: Optional[U
 
     try:
         if not is_saved:
-            response = await client.post(
+            response = await aeros_post(
-            f"{AEROS_BASE_URL}/api/Simulation/RunSimulation",
+                "/api/Simulation/RunSimulation",
-            json=sim_data,
+                json=sim_data,
-            headers={"Content-Type": "application/json"},
+                headers={"Content-Type": "application/json"},
-        )
+            )
             response.raise_for_status()
             result = response.json()
             
@@ -420,17 +420,15 @@ async def execute_simulation(*, db_session: DbSession, simulation_id: Optional[U
         print("Simulation started with id: %s", simulation.id)
         
         # if not os.path.exists(tmpfile):
-        response = requests.post(f"{AEROS_BASE_URL}/api/Simulation/RunSimulation", stream=True, json=sim_data)
+        # Using aeros_post wrapper which uses LicensedSession
+        response = await aeros_post(
+            "/api/Simulation/RunSimulation",
+            stream=True,
+            json=sim_data
+        )
         file_obj = response.raw
                         
 
-        # response = await client.post(
-        #     f"{AEROS_BASE_URL}/api/Simulation/RunSimulation",
-        #     json=sim_data,
-        #     headers={"Content-Type": "application/json"},
-        # )
-        # response.raise_for_status()
-        # result = response.json()
 
         await save_simulation_result(
             db_session=db_session, simulation_id=simulation.id, schematic_name=sim_data["SchematicName"], eq_update=eq_update, file_path=file_obj

src/aeros_simulation/simulation_save_service.py

@@ -17,7 +17,7 @@ from fastapi import HTTPException, status
 from src.aeros_simulation.model import AerosSimulationCalcResult, AerosSimulationPlotResult
 from src.aeros_simulation.service import get_all_aeros_node, get_or_save_node, get_plant_calc_result, get_simulation_by_id, get_simulation_with_plot_result
 from src.aeros_simulation.utils import calculate_eaf, calculate_eaf_konkin
-from src.config import AEROS_BASE_URL
+from src.aeros_utils import aeros_post
 from src.database.core import DbSession
 
 
@@ -32,17 +32,16 @@ async def execute_simulation(
     tmpfile = os.path.join(tempfile.gettempdir(), f"simulation_{simulation_id}.json")
 
     try:
-        async with httpx.AsyncClient(timeout=None) as client:
+        response = await aeros_post(
-            async with client.stream(
+            "/api/Simulation/RunSimulation",
-                "POST",
+            json=sim_data,
-                f"{AEROS_BASE_URL}/api/Simulation/RunSimulation",
+            headers={"Content-Type": "application/json"},
-                json=sim_data,
+            stream=True
-                headers={"Content-Type": "application/json"},
+        )
-            ) as response:
+        response.raise_for_status()
-                response.raise_for_status()
+        with open(tmpfile, "wb") as f:
-                with open(tmpfile, "wb") as f:
+            for chunk in response.iter_content(chunk_size=8192):
-                    async for chunk in response.aiter_bytes():
+                f.write(chunk)
-                        f.write(chunk)
 
         if not is_saved:
             # If not saving to DB, just return parsed JSON

src/aeros_utils.py

@@ -0,0 +1,32 @@
+import anyio
+from licaeros import LicensedSession, device_fingerprint_hex
+from src.config import AEROS_BASE_URL, AEROS_LICENSE_ID, AEROS_LICENSE_SECRET
+import logging
+
+log = logging.getLogger(__name__)
+
+# Initialize a global session if possible, or create on demand
+_aeros_session = None
+
+def get_aeros_session():
+    global _aeros_session
+    if _aeros_session is None:
+        log.info(f"Initializing LicensedSession with base URL: {AEROS_BASE_URL}")
+        log.info(f"Encrypted Device ID: {device_fingerprint_hex()}")
+        _aeros_session = LicensedSession(
+            api_base=AEROS_BASE_URL,
+            license_id=AEROS_LICENSE_ID,
+            license_secret=AEROS_LICENSE_SECRET,
+        )
+    return _aeros_session
+
+async def aeros_post(path: str, json: dict = None, **kwargs):
+    """
+    Asynchronous wrapper for LicensedSession.post
+    """
+    session = get_aeros_session()
+    # LicensedSession might not be async-compatible, so we run it in a thread
+    response = await anyio.to_thread.run_sync(
+        lambda: session.post(path, json)
+    )
+    return response

src/api.py

@@ -21,13 +21,6 @@ class ErrorResponse(BaseModel):
 
 api_router = APIRouter(
     default_response_class=JSONResponse,
-    responses={
-        400: {"model": ErrorResponse},
-        401: {"model": ErrorResponse},
-        403: {"model": ErrorResponse},
-        404: {"model": ErrorResponse},
-        500: {"model": ErrorResponse},
-    },
 )
 
 

src/config.py

@@ -96,4 +96,7 @@ TEMPORAL_URL = config("TEMPORAL_URL", default="http://192.168.1.86:7233")
 RELIABILITY_SERVICE_API = config("RELIABILITY_SERVICE_API", default="http://192.168.1.82:8000/reliability")
 
 CLAMAV_HOST = config("CLAMAV_HOST", default="192.168.1.82")
-CLAMAV_PORT = config("CLAMAV_PORT", cast=int, default=3310)
+CLAMAV_PORT = config("CLAMAV_PORT", cast=int, default=3310)
+
+AEROS_LICENSE_ID = config("AEROS_LICENSE_ID", default="")
+AEROS_LICENSE_SECRET = config("AEROS_LICENSE_SECRET", default="")

src/exceptions.py

@@ -5,6 +5,7 @@ from typing import Any, Dict, List, Optional
 from asyncpg.exceptions import DataError as AsyncPGDataError
 from asyncpg.exceptions import PostgresError
 from fastapi import FastAPI, HTTPException, Request
+from starlette.exceptions import HTTPException as StarletteHTTPException
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel, Field
@@ -14,6 +15,7 @@ from sqlalchemy.exc import DataError, DBAPIError, IntegrityError, SQLAlchemyErro
 
 from src.enums import ResponseStatus
 
+log = logging.getLogger(__name__)
 
 class ErrorDetail(BaseModel):
     field: Optional[str] = Field(None, max_length=100)
@@ -56,7 +58,7 @@ def get_request_context(request: Request):
 
     return {
         "endpoint": request.url.path,
-        "url": request.url,
+        "url": str(request.url),
         "method": request.method,
         "remote_addr": get_client_ip(),
     }
@@ -102,30 +104,63 @@ def handle_exception(request: Request, exc: Exception):
     Global exception handler for Fastapi application.
     """
     request_info = get_request_context(request)
+    
 
     if isinstance(exc, RateLimitExceeded):
-        _rate_limit_exceeded_handler(request, exc)
+        return _rate_limit_exceeded_handler(request, exc)
-    if isinstance(exc, HTTPException):
+    if isinstance(exc, RequestValidationError):
-        logging.error(
+        log.error(
-            f"HTTP exception | Code: {exc.status_code} | Error: {exc.detail} | Request: {request_info}",
+            "Validation error occurred",
-            extra={"error_category": "http"},
+            extra={
+                "error_category": "validation",
+                "errors": exc.errors(),
+                "request": request_info,
+            },
+        )
+        return JSONResponse(
+            status_code=422,
+            content={
+                "data": None,
+                "message": "Validation Error",
+                "status": ResponseStatus.ERROR,
+                "errors": exc.errors(),
+            },
+        )
+    if isinstance(exc, (HTTPException, StarletteHTTPException)):
+        log.error(
+            "HTTP exception occurred",
+            extra={
+                "error_category": "http",
+                "status_code": exc.status_code,
+                "detail": exc.detail if hasattr(exc, "detail") else str(exc),
+                "request": request_info,
+            },
         )
 
         return JSONResponse(
             status_code=exc.status_code,
             content={
                 "data": None,
-                "message": str(exc.detail),
+                "message": str(exc.detail) if hasattr(exc, "detail") else str(exc),
                 "status": ResponseStatus.ERROR,
-                "errors": [ErrorDetail(message=str(exc.detail)).model_dump()],
+                "errors": [
+                    ErrorDetail(
+                        message=str(exc.detail) if hasattr(exc, "detail") else str(exc)
+                    ).model_dump()
+                ],
             },
         )
 
     if isinstance(exc, SQLAlchemyError):
         error_message, status_code = handle_sqlalchemy_error(exc)
-        logging.error(
+        log.error(
-            f"Database Error | Error: {str(error_message)} | Request: {request_info}",
+            "Database error occurred",
-            extra={"error_category": "database"},
+            extra={
+                "error_category": "database",
+                "error_message": error_message,
+                "request": request_info,
+                "exception": str(exc),
+            },
         )
 
         return JSONResponse(
@@ -139,9 +174,14 @@ def handle_exception(request: Request, exc: Exception):
         )
 
     # Log unexpected errors
-    logging.error(
+    log.error(
-        f"Unexpected Error | Error: {str(exc)} | Request: {request_info}",
+        "Unexpected error occurred",
-        extra={"error_category": "unexpected"},
+        extra={
+            "error_category": "unexpected",
+            "error_message": str(exc),
+            "request": request_info,
+        },
+        exc_info=True,
     )
 
     return JSONResponse(

src/logging.py

@@ -117,13 +117,17 @@ def configure_logging():
     handler.setFormatter(formatter)
     root_logger.addHandler(handler)
 
-    # Reconfigure uvicorn loggers to use our JSON formatter
+    for logger_name in ["uvicorn", "uvicorn.error", "fastapi"]:
-    for logger_name in ["uvicorn", "uvicorn.access", "uvicorn.error", "fastapi"]:
         logger = logging.getLogger(logger_name)
         logger.handlers = []
         logger.propagate = True
 
-    # sometimes the slack client can be too verbose
+    for logger_name in ["uvicorn.access"]:
-    logging.getLogger("slack_sdk.web.base_client").setLevel(logging.CRITICAL)
+        logger = logging.getLogger(logger_name)
+        logger.handlers = []
+        logger.propagate = False
+
+    # # sometimes the slack client can be too verbose
+    # logging.getLogger("slack_sdk.web.base_client").setLevel(logging.CRITICAL)
 
     

src/main.py

@@ -7,6 +7,7 @@ from uuid import uuid1
 from typing import Optional, Final
 
 from fastapi import FastAPI, HTTPException, status, Path
+from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
 from pydantic import ValidationError
 
@@ -32,21 +33,29 @@ from src.database.core import engine, async_session, async_aeros_session
 from src.exceptions import handle_exception
 from src.middleware import RequestValidationMiddleware
 from src.context import set_request_id, reset_request_id, get_request_id
+from sqlalchemy.exc import SQLAlchemyError
 
 log = logging.getLogger(__name__)
 
+from starlette.exceptions import HTTPException as StarletteHTTPException
+
 # we configure the logging level and format
 configure_logging()
 
-# we define the exception handlers
-exception_handlers = {Exception: handle_exception}
-
 # we create the ASGI for the app
-app = FastAPI(exception_handlers=exception_handlers, openapi_url="", title="LCCA API",
+app = FastAPI(openapi_url="", title="LCCA API",
               description="Welcome to RBD's API documentation!",
               version="0.1.0")
-app.state.limiter = limiter
+
+# we define the exception handlers
+app.add_exception_handler(Exception, handle_exception)
+app.add_exception_handler(HTTPException, handle_exception)
+app.add_exception_handler(StarletteHTTPException, handle_exception)
+app.add_exception_handler(RequestValidationError, handle_exception)
 app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+app.add_exception_handler(SQLAlchemyError, handle_exception)
+
+app.state.limiter = limiter
 app.add_middleware(GZipMiddleware, minimum_size=2000)
 
 app.add_middleware(RequestValidationMiddleware)
@@ -72,11 +81,25 @@ async def db_session_middleware(request: Request, call_next):
         process_time = (time.time() - start_time) * 1000
         
         log.info(
-            f"Request: {request.method} {request.url.path} Status: {response.status_code} Duration: {process_time:.2f}ms"
+            "Request finished",
+            extra={
+                "method": request.method,
+                "path": request.url.path,
+                "status_code": response.status_code,
+                "duration_ms": round(process_time, 2),
+            },
         )
-            
+
     except Exception as e:
-        log.error(f"Request failed: {request.method} {request.url.path} Error: {str(e)}")
+        log.error(
+            "Request failed",
+            extra={
+                "method": request.method,
+                "path": request.url.path,
+                "error": str(e),
+            },
+            exc_info=True,
+        )
         raise e from None
     finally:
         await request.state.db.close()