paginate the top 10 equipments service

src/database/service.py

@@ -77,6 +77,36 @@ def search(*, query_str: str, query: Query, model, sort=False):
     return query.params(term=query_str)
 
 
+def _extract_result_items(result):
+    """Normalize SQLAlchemy result rows into ORM entities with labeled columns attached."""
+
+    rows = result.fetchall()
+    items = []
+
+    for row in rows:
+        row_values = tuple(row)
+        if not row_values:
+            continue
+
+        primary = row_values[0]
+
+        if hasattr(primary, "__table__"):
+            mapping = getattr(row, "_mapping", None)
+            if mapping:
+                for key, value in mapping.items():
+                    if isinstance(key, str):
+                        setattr(primary, key, value)
+            items.append(primary)
+        else:
+            # Fall back to first column (scalar or tuple) if no ORM entity is present.
+            if len(row_values) == 1:
+                items.append(primary)
+            else:
+                items.append(row_values)
+
+    return items
+
+
 async def search_filter_sort_paginate(
     db_session: DbSession,
     model,
@@ -106,7 +136,7 @@ async def search_filter_sort_paginate(
     total = await db_session.scalar(count_query)
     if all:
         result = await db_session.execute(query)
-        items = result.scalars().all()
+        items = _extract_result_items(result)
         return {
             "items": items,
             "itemsPerPage": total,
@@ -118,7 +148,7 @@ async def search_filter_sort_paginate(
     query = query.offset((page - 1) * items_per_page).limit(items_per_page)
 
     result = await db_session.execute(query)
-    items = result.scalars().all()
+    items = _extract_result_items(result)
 
     # try:
     #     query, pagination = apply_pagination(

src/equipment/router.py

@@ -9,7 +9,8 @@ from src.equipment.schema import (
     EquipmentPagination,
     EquipmentRead,
     EquipmentCreate,
-    EquipmentTop10EconomicLife,
+    EquipmentTop10,
+    EquipmentTop10Pagination,
     EquipmentUpdate,
 )
 from src.equipment.service import (
@@ -115,10 +116,10 @@ async def simulate_equipment(db_session: DbSession, assetnum: str):
 
 @router.get(
     "/top-10-replacement-priorities",
-    response_model=StandardResponse[List[EquipmentTop10EconomicLife]],
+    response_model=StandardResponse[EquipmentTop10Pagination],
 )
-async def get_calculated_top_10_replacement_priorities(db_session: DbSession):
+async def get_calculated_top_10_replacement_priorities(db_session: DbSession, common: CommonParameters):
-    equipment_data = await get_top_10_replacement_priorities(db_session=db_session)
+    equipment_data = await get_top_10_replacement_priorities(db_session=db_session, common=common)
     return StandardResponse(
         data=equipment_data,
         message="Top 10 Replacement Priorities Data retrieved successfully",
@@ -126,10 +127,10 @@ async def get_calculated_top_10_replacement_priorities(db_session: DbSession):
 
 @router.get(
     "/top-10-economic-life",
-    response_model=StandardResponse[List[EquipmentTop10EconomicLife]],
+    response_model=StandardResponse[EquipmentTop10Pagination],
 )
-async def get_calculated_top_10_economic_life(db_session: DbSession):
+async def get_calculated_top_10_economic_life(db_session: DbSession, common: CommonParameters):
-    equipment_data = await get_top_10_economic_life(db_session=db_session)
+    equipment_data = await get_top_10_economic_life(db_session=db_session, common=common)
     return StandardResponse(
         data=equipment_data,
         message="Top 10 Economic Life Data retrieved successfully",

src/equipment/schema.py

@@ -108,7 +108,7 @@ class EquipmentRead(DefaultBase):
     maximo_data: Optional[List[dict]] = Field(None, nullable=True)
 
 
-class EquipmentTop10EconomicLife(EquipmentBase):
+class EquipmentTop10(EquipmentBase):
     id: UUID
     equipment_master: EquipmentMasterBase
     forecasting_target_year: Optional[int] = Field(None, nullable=True)
@@ -123,6 +123,8 @@ class EquipmentTop10EconomicLife(EquipmentBase):
     harga_saat_ini: Optional[float] = Field(None, nullable=True, le=MAX_PRICE)
     economic_life: Optional[int] = Field(None, nullable=True)
 
+class EquipmentTop10Pagination(Pagination):
+    items: List[EquipmentTop10] = []
 
 class EquipmentDataMaster(EquipmentBase):
     id: UUID

src/equipment/service.py

@@ -338,7 +338,7 @@ async def get_all(
     return result
 
 
-async def get_top_10_economic_life(*, db_session: DbSession) -> list[Equipment]:
+async def get_top_10_economic_life(*, db_session: DbSession, common) -> list[Equipment]:
     """Returns top 10 economic life."""
     query = (
         Select(Equipment)
@@ -354,20 +354,13 @@ async def get_top_10_economic_life(*, db_session: DbSession) -> list[Equipment]:
         )
         .filter(Equipment.minimum_eac_year != None)
         .order_by(func.abs(current_year - Equipment.minimum_eac_year).desc())
-        .limit(10)
     )
 
-    result = await db_session.execute(query)
+    # result = await db_session.execute(query)
-    
+    result = await search_filter_sort_paginate(model=query, **common)
-    equipment_list = []
+    return result
-    for row in result.all():
-        equipment = row[0]
-        equipment.economic_life = row[1]
-        equipment_list.append(equipment)
-
-    return equipment_list
 
-async def get_top_10_replacement_priorities(*, db_session: DbSession) -> list[Equipment]:
+async def get_top_10_replacement_priorities(*, db_session: DbSession, common) -> list[Equipment]:
     """Returns top 10 replacement priorities."""
     query = (
         Select(Equipment)
@@ -384,18 +377,11 @@ async def get_top_10_replacement_priorities(*, db_session: DbSession) -> list[Eq
         .filter(Equipment.minimum_eac_year != None)
         .order_by(func.abs(current_year - Equipment.minimum_eac_year).asc())
         .order_by(func.abs(Equipment.minimum_eac).desc())
-        .limit(10)
     )
 
-    result = await db_session.execute(query)
+    # result = await db_session.execute(query)
-    
+    result = await search_filter_sort_paginate(model=query, **common)
-    equipment_list = []
+    return result
-    for row in result.all():
-        equipment = row[0]
-        equipment.economic_life = row[1]
-        equipment_list.append(equipment)
-
-    return equipment_list
 
 async def generate_all_transaction(*, db_session: DbSession, token):
     """Generate transaction for all equipments in the database based on equipments assetnum."""