add pagination to master equipment api

.gitignore

@@ -1,2 +1,4 @@
 env/
-.env
+.env
+
+__pycache__/

src/database/service.py

@@ -91,7 +91,11 @@ async def search_filter_sort_paginate(
 ):
     """Common functionality for searching, filtering, sorting, and pagination."""
     # try:
-    query = Select(model)
+
+    if not isinstance(model, Select):
+        query = Select(model)
+    else:
+        query = model
 
     if query_str:
         sort = False if sort_by else True
@@ -101,13 +105,13 @@ async def search_filter_sort_paginate(
     # Get total count
     count_query = Select(func.count()).select_from(query.subquery())
     total = await db_session.scalar(count_query)
-    
+
     query = (
         query
         .offset((page - 1) * items_per_page)
         .limit(items_per_page)
     )
-    
+
     result = await db_session.execute(query)
     items = result.scalars().all()
 
@@ -126,6 +130,7 @@ async def search_filter_sort_paginate(
     return {
         "items": items,
         "itemsPerPage": items_per_page,
+        "totalPages": (total + items_per_page - 1) // items_per_page,
         "page": page,
         "total": total,
     }

src/equipment_master/router.py

@@ -3,19 +3,36 @@
 from typing import Annotated, List, Optional
 
 from src.models import StandardResponse
-from .service import get_all_master
+from .service import get_all_master, get_master
-from fastapi import APIRouter, Query
+from fastapi import APIRouter, HTTPException, Query, status
-from .schema import EquipmentMasterTree
+from .schema import EquipmentMasterPaginated, EquipmentMasterRead
+from src.database.service import search_filter_sort_paginate, CommonParameters
 from src.database.core import DbSession
 router = APIRouter()
 
 
-@router.get("", response_model=StandardResponse[List[EquipmentMasterTree]])
+@router.get("", response_model=StandardResponse[EquipmentMasterPaginated])
 async def get_all_equipment_master_tree(
     db_session: DbSession,
+    common: CommonParameters,
     parent_id: Annotated[Optional[str], Query(description="Parent ID")] = None,
-):
-    equipment_masters = await get_all_master(parent_id=parent_id, db_session=db_session)
 
+):
+    equipment_masters = await get_all_master(parent_id=parent_id, db_session=db_session, common=common)
 
     return StandardResponse(data=equipment_masters, message="Data retrieved successfully")
+
+
+@router.get("/{equipment_master_id}", response_model=StandardResponse[EquipmentMasterRead])
+async def get_equipment_master_tree(
+    db_session: DbSession, equipment_master_id: str
+):
+    equipment_master = await get_master(db_session=db_session, equipment_master_id=equipment_master_id)
+
+    if not equipment_master:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="A data with this id does not exist.",
+        )
+
+    return StandardResponse(data=equipment_master, message="Data retrieved successfully")

src/equipment_master/schema.py

@@ -1,6 +1,6 @@
 
 from datetime import datetime
-from typing import List, Optional
+from typing import ForwardRef, List, Optional
 from uuid import UUID
 
 from pydantic import Field
@@ -21,6 +21,7 @@ class EquipmentMasterCreate(EquipmentMasterBase):
 class EquipmentMasterUpdate(EquipmentMasterBase):
     pass
 
+EquipmentMasterReadRef = ForwardRef('EquipmentMasterRead')
 
 class EquipmentMasterRead(EquipmentMasterBase):
     id: UUID
@@ -29,7 +30,9 @@ class EquipmentMasterRead(EquipmentMasterBase):
     system_tag: Optional[str] = Field(None, nullable=True)
     assetnum: Optional[str] = Field(None, nullable=True)
     location_tag: Optional[str] = Field(None, nullable=True)
+    children: List[EquipmentMasterReadRef] # type: ignore
 
 
-class EquipmentMasterTree(EquipmentMasterRead):
+
-    children: List[EquipmentMasterRead] = []
+class EquipmentMasterPaginated(Pagination):
+    items: List[EquipmentMasterRead] = []

src/equipment_master/service.py

@@ -5,12 +5,25 @@ from src.database.core import DbSession
 from sqlalchemy.orm import selectinload
 from typing import Optional
 from .model import EquipmentMaster
+from src.database.service import search_filter_sort_paginate
+from sqlalchemy.orm import Load
 
 
-async def get_all_master(*, db_session: DbSession, parent_id: Optional[str] = None):
+def recursive_load(depth=5):
-    query = Select(EquipmentMaster).options(selectinload(EquipmentMaster.children))
+    loader = Load(EquipmentMaster)
-    results = await db_session.execute(query)
+    for _ in range(depth):
-    results = results.unique().scalars().all()
+        loader = loader.selectinload(EquipmentMaster.children)
+    return loader
+
+
+async def get_all_master(*, db_session: DbSession, parent_id: Optional[str] = None, common):
+    query = Select(EquipmentMaster).filter(EquipmentMaster.parent_id ==
+                                           parent_id).options(recursive_load(5))
+
+    results = await search_filter_sort_paginate(model=query, **common)
+    # results = await db_session.execute(query)
+    # results = results.unique().scalars().all()
+
     return results
 
 
@@ -18,9 +31,6 @@ async def get_master(*, db_session: DbSession, equipment_master_id: str):
     query = Select(EquipmentMaster).filter(
         EquipmentMaster.id == equipment_master_id)
     result = await db_session.execute(query)
-    result = result.scalars().one_or_none()
+    result = result.unique().scalars().one_or_none()
-
-    if result:
-        await db_session.refresh(result)
 
     return result

src/main.py

@@ -48,13 +48,6 @@ app = FastAPI(exception_handlers=exception_handlers, openapi_url="", title="LCCA
 app.state.limiter = limiter
 app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 app.add_middleware(GZipMiddleware, minimum_size=2000)
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=['*'],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
 
 
 REQUEST_ID_CTX_KEY: Final[str] = "request_id"