add parts

src/scope_equipment_part/router.py

@@ -1,4 +1,5 @@
 
+from typing import Dict, List
 from fastapi import APIRouter, HTTPException, Query, status
 
 
@@ -11,13 +12,11 @@ from src.database.service import CommonParameters, search_filter_sort_paginate,
 router = APIRouter()
 
 
-@router.get("/{assetnum}", response_model=StandardResponse[ScopeEquipmentActivityPagination])
+@router.get("/{assetnum}", response_model=StandardResponse[List[Dict]])
-async def get_scope_equipment_activities(common: CommonParameters, assetnum):
+async def get_scope_equipment_parts(db_session: DbSession, assetnum):
     """Get all scope activity pagination."""
     # return
-    data = await get_all(common=common, assetnum=assetnum)
+    data = await get_all(db_session=db_session, assetnum=assetnum)
-    
-    raise Exception(data)
 
     return StandardResponse(
         data=data,

src/scope_equipment_part/service.py

@@ -1,5 +1,6 @@
 
 
+import random
 from sqlalchemy import Select, Delete, and_
 from sqlalchemy.orm import selectinload
 from typing import Optional
@@ -20,37 +21,62 @@ from src.auth.service import CurrentUser
 #     result = await db_session.get(ScopeEquipmentActivity, scope_equipment_activity_id)
 #     return result
 
-
+def create_dummy_parts(assetnum: str, count: int = 5):
-async def get_all(common: CommonParameters, assetnum: Optional[str]):
+    """
-    if not assetnum:
+    Create a list of dummy ScopeEquipmentPart objects with random stock values.
-        raise ValueError("assetnum parameter is required")
+
-
+    Args:
-    db_session: DbSession = common.get("db_session")
+        assetnum (str): The base asset number to generate dummy parts for.
-
+        count (int): The number of parts to create. Default is 5.
-    # First get the parent equipment
+
-    equipment_stmt = Select(MasterEquipment).where(
+    Returns:
-        MasterEquipment.assetnum == assetnum)
+        List[ScopeEquipmentPart]: A list of dummy ScopeEquipmentPart objects.
-    equipment: MasterEquipment = await db_session.scalar(equipment_stmt)
+    """
-
+    parts = []
-    if not equipment:
+    for i in range(1, count + 1):
-        raise ValueError(f"No equipment found with assetnum: {assetnum}")
+        # Generate a unique part asset number
-
+        part_assetnum = f"{assetnum}_PART_{i}"
-    # Build query for parts
+        stock = random.randint(1, 100)  # Random stock value between 1 and 100
-    stmt = (
+        parts.append({
-        Select(ScopeEquipmentPart)
+            "assetnum": part_assetnum,
-        .join(ScopeEquipmentPart.master_equipments)
+            "stock": stock
-        .join(MasterEquipment.equipment_tree)
+        })
-        .where(
+    return parts
-            and_(
+
-                MasterEquipment.parent_id == equipment.id,
+
-                MasterEquipmentTree.level_no == 4
+async def get_all(db_session: DbSession, assetnum: Optional[str]):
-            )
+    # Example usage
-        ).options(selectinload(ScopeEquipmentPart.master_equipments))
+    dummy_parts = create_dummy_parts(assetnum, count=10)
-    )
+    
-
+    # if not assetnum:
-    results = await search_filter_sort_paginate(model=stmt, **common)
+    #     raise ValueError("assetnum parameter is required")
-
+
-    return results
+    # db_session: DbSession = common.get("db_session")
+
+    # # First get the parent equipment
+    # equipment_stmt = Select(MasterEquipment).where(
+    #     MasterEquipment.assetnum == assetnum)
+    # equipment: MasterEquipment = await db_session.scalar(equipment_stmt)
+
+    # if not equipment:
+    #     raise ValueError(f"No equipment found with assetnum: {assetnum}")
+
+    # # Build query for parts
+    # stmt = (
+    #     Select(ScopeEquipmentPart)
+    #     .join(ScopeEquipmentPart.master_equipments)
+    #     .join(MasterEquipment.equipment_tree)
+    #     .where(
+    #         and_(
+    #             MasterEquipment.parent_id == equipment.id,
+    #             MasterEquipmentTree.level_no == 4
+    #         )
+    #     ).options(selectinload(ScopeEquipmentPart.master_equipments))
+    # )
+
+    # results = await search_filter_sort_paginate(model=stmt, **common)
+
+    return dummy_parts
 
 
 # async def create(*, db_session: DbSession, scope_equipment_activty_in: ScopeEquipmentActivityCreate):