add validation

src/acquisition_cost/router.py

@@ -1,14 +1,15 @@
-from typing import Optional
-from fastapi import APIRouter, HTTPException, status, Query
+from typing import Annotated, Optional
+from fastapi import APIRouter, Depends, HTTPException, status, Query
+from pydantic import ValidationError
 
 from src.acquisition_cost.model import AcquisitionData
-from src.acquisition_cost.schema import AcquisitionCostDataPagination, AcquisitionCostDataRead, AcquisitionCostDataCreate, AcquisitionCostDataUpdate
+from src.acquisition_cost.schema import AcquisitionCostDataPagination, AcquisitionCostDataRead, AcquisitionCostDataCreate, AcquisitionCostDataUpdate, ListQueryParams
 from src.acquisition_cost.service import get, get_all, create, update, delete
 
 from src.database.service import CommonParameters, search_filter_sort_paginate
 from src.database.core import DbSession
 from src.auth.service import CurrentUser
-from src.models import StandardResponse
+from src.models import CommonParams, StandardResponse
 
 router = APIRouter()
 
@@ -17,14 +18,13 @@ router = APIRouter()
 async def get_yeardatas(
     db_session: DbSession,
     common: CommonParameters,
-    items_per_page: Optional[int] = Query(5),
-    search: Optional[str] = Query(None),
+    params: Annotated[ListQueryParams, Query()],
 ):
     """Get all acquisition_cost_data pagination."""    
     get_acquisition_cost_data = await get_all(
         db_session=db_session,
-        items_per_page=items_per_page,
-        search=search,
+        items_per_page=params.items_per_page,
+        search=params.search,
         common=common,
     )
     # return

src/acquisition_cost/schema.py

@@ -3,7 +3,7 @@ from typing import List, Optional
 from uuid import UUID
 
 from pydantic import Field
-from src.models import DefaultBase, Pagination
+from src.models import CommonParams, DefaultBase, Pagination
 
 
 class AcquisitionCostDataBase(DefaultBase):
@@ -31,3 +31,16 @@ class AcquisitionCostDataRead(AcquisitionCostDataBase):
 
 class AcquisitionCostDataPagination(Pagination):
     items: List[AcquisitionCostDataRead] = []
+
+
+class ListQueryParams(CommonParams):
+    items_per_page: Optional[int] = Field(
+        default=5,
+        ge=1,
+        le=1000,
+        description="Number of items per page"
+    )
+    search: Optional[str] = Field(
+        default=None,
+        description="Search keyword"
+    )

src/equipment/router.py

@@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import Annotated, List, Optional
 from fastapi import APIRouter, HTTPException, status, Query
 from fastapi.responses import StreamingResponse
 import json
@@ -13,6 +13,7 @@ from src.equipment.schema import (
     EquipmentTop10Pagination,
     EquipmentUpdate,
     CountRemainingLifeResponse,
+    ListQueryParams,
 )
 from src.equipment.service import (
     get_master_by_assetnum,
@@ -34,7 +35,7 @@ from src.modules.equipment.Eac import Eac
 from src.database.service import CommonParameters, search_filter_sort_paginate
 from src.database.core import DbSession, CollectorDbSession
 from src.auth.service import CurrentUser, Token
-from src.models import StandardResponse
+from src.models import CommonParams, StandardResponse
 
 router = APIRouter()
 
@@ -43,15 +44,14 @@ router = APIRouter()
 async def get_equipments(
     db_session: DbSession,
     common: CommonParameters,
-    items_per_page: Optional[int] = Query(5),
-    search: Optional[str] = Query(None),
+    params: Annotated[ListQueryParams, Query()],
 ):
     """Get all equipment pagination."""
     # return
     equipment_data = await get_all(
         db_session=db_session,
-        items_per_page=items_per_page,
-        search=search,
+        items_per_page=params.items_per_page,
+        search=params.search,
         common=common,
     )
     return StandardResponse(
@@ -119,7 +119,7 @@ async def simulate_equipment(db_session: DbSession, assetnum: str):
     "/count-remaining-life",
     response_model=StandardResponse[CountRemainingLifeResponse],
 )
-async def get_count_remaining_life_equipment(db_session: DbSession, common: CommonParameters):
+async def get_count_remaining_life_equipment(db_session: DbSession, common: CommonParameters, params: Annotated[CommonParams, Query()]):
     count = await get_count_remaining_life(db_session=db_session, common=common)
     return StandardResponse(
         data=count,
@@ -130,7 +130,7 @@ async def get_count_remaining_life_equipment(db_session: DbSession, common: Comm
     "/top-10-replacement-priorities",
     response_model=StandardResponse[EquipmentTop10Pagination],
 )
-async def get_calculated_top_10_replacement_priorities(db_session: DbSession, common: CommonParameters):
+async def get_calculated_top_10_replacement_priorities(db_session: DbSession, common: CommonParameters, params: Annotated[CommonParams, Query()]):
     equipment_data = await get_top_10_replacement_priorities(db_session=db_session, common=common)
     return StandardResponse(
         data=equipment_data,
@@ -141,7 +141,7 @@ async def get_calculated_top_10_replacement_priorities(db_session: DbSession, co
     "/top-10-economic-life",
     response_model=StandardResponse[EquipmentTop10Pagination],
 )
-async def get_calculated_top_10_economic_life(db_session: DbSession, common: CommonParameters):
+async def get_calculated_top_10_economic_life(db_session: DbSession, common: CommonParameters, params: Annotated[CommonParams, Query()]):
     equipment_data = await get_top_10_economic_life(db_session=db_session, common=common)
     return StandardResponse(
         data=equipment_data,

src/equipment/schema.py

@@ -3,7 +3,7 @@ from typing import List, Optional
 from uuid import UUID
 
 from pydantic import Field
-from src.models import DefaultBase, Pagination
+from src.models import CommonParams, DefaultBase, Pagination
 
 MAX_PRICE = 1_000_000_000_000_000  # thousands of trillion
 
@@ -150,3 +150,16 @@ class CountRemainingLifeResponse(DefaultBase):
     safe: int
     warning: int
     critical: int
+
+class ListQueryParams(CommonParams):
+    items_per_page: Optional[int] = Field(
+        default=5,
+        ge=1,
+        le=1000,
+        description="Number of items per page"
+    )
+    search: Optional[str] = Field(
+        default=None,
+        description="Search keyword"
+    )
+

src/equipment_master/router.py

@@ -3,7 +3,7 @@ from typing import Annotated, List, Optional
 from src.models import StandardResponse
 from .service import get_all_master, get_master
 from fastapi import APIRouter, HTTPException, Query, status
-from .schema import EquipmentMasterPaginated, EquipmentMasterRead
+from .schema import EquipmentMasterPaginated, EquipmentMasterQuery, EquipmentMasterRead
 from src.database.service import search_filter_sort_paginate, CommonParameters
 from src.database.core import DbSession
 
@@ -14,14 +14,12 @@ router = APIRouter()
 async def get_all_equipment_master_tree(
     db_session: DbSession,
     common: CommonParameters,
-    parent_id: Annotated[Optional[str], Query(description="Parent ID")] = None,
-    search: Optional[str] = Query(None),
-    items_per_page: Optional[int] = Query(5),
+    params: Annotated[EquipmentMasterQuery, Query()],
 ):
     equipment_masters = await get_all_master(
-        parent_id=parent_id,
-        search=search,
-        items_per_page=items_per_page,
+        parent_id=params.parent_id,
+        search=params.search,
+        items_per_page=params.items_per_page,
         db_session=db_session,
         common=common,
     )

src/equipment_master/schema.py

@@ -3,7 +3,7 @@ from typing import ForwardRef, List, Optional
 from uuid import UUID
 
 from pydantic import Field
-from src.models import DefaultBase, Pagination
+from src.models import CommonParams, DefaultBase, Pagination
 
 
 class EquipmentMasterBase(DefaultBase):
@@ -42,3 +42,9 @@ class EquipmentMasterRead(EquipmentMasterBase):
 
 class EquipmentMasterPaginated(Pagination):
     items: List[EquipmentMasterRead] = []
+
+
+class EquipmentMasterQuery(CommonParams):
+    parent_id : Optional[str] = None
+    items_per_page : Optional[int] = 5
+    search : Optional[str] = None

src/manpower_cost/router.py

@@ -1,8 +1,8 @@
-from typing import Optional
+from typing import Annotated, Optional
 from fastapi import APIRouter, HTTPException, status, Query
 
 from src.manpower_cost.model import ManpowerCost
-from src.manpower_cost.schema import ManpowerCostPagination, ManpowerCostRead, ManpowerCostCreate, ManpowerCostUpdate
+from src.manpower_cost.schema import ManpowerCostPagination, ManpowerCostRead, ManpowerCostCreate, ManpowerCostUpdate, QueryParams
 from src.manpower_cost.service import get, get_all, create, update, delete
 
 from src.database.service import CommonParameters, search_filter_sort_paginate
@@ -17,14 +17,13 @@ router = APIRouter()
 async def get_yeardatas(
     db_session: DbSession,
     common: CommonParameters,
-    items_per_page: Optional[int] = Query(5),
-    search: Optional[str] = Query(None),
+    params: Annotated[QueryParams, Query()],
 ):
     """Get all acquisition_cost_data pagination."""
     get_acquisition_cost_data = await get_all(
         db_session=db_session,
-        items_per_page=items_per_page,
-        search=search,
+        items_per_page=params.items_per_page,
+        search=params.search,
         common=common,
     )
     # return

src/manpower_cost/schema.py

@@ -3,7 +3,7 @@ from typing import List, Optional
 from uuid import UUID
 
 from pydantic import Field
-from src.models import DefaultBase, Pagination
+from src.models import CommonParams, DefaultBase, Pagination
 
 
 class ManpowerCostBase(DefaultBase):
@@ -31,3 +31,8 @@ class ManpowerCostRead(ManpowerCostBase):
 
 class ManpowerCostPagination(Pagination):
     items: List[ManpowerCostRead] = []
+
+
+class QueryParams(CommonParams):
+    items_per_page: Optional[int] = Field(5)
+    search: Optional[str] = Field(None)

src/manpower_master/router.py

@@ -1,8 +1,8 @@
-from typing import Optional
+from typing import Annotated, Optional
 from fastapi import APIRouter, HTTPException, status, Query
 
 from src.manpower_cost.model import ManpowerCost
-from src.manpower_cost.schema import ManpowerCostPagination, ManpowerCostRead, ManpowerCostCreate, ManpowerCostUpdate
+from src.manpower_cost.schema import ManpowerCostPagination, ManpowerCostRead, ManpowerCostCreate, ManpowerCostUpdate, QueryParams
 from src.manpower_cost.service import get, get_all, create, update, delete
 
 from src.database.service import CommonParameters, search_filter_sort_paginate
@@ -17,14 +17,13 @@ router = APIRouter()
 async def get_yeardatas(
     db_session: DbSession,
     common: CommonParameters,
-    items_per_page: Optional[int] = Query(5),
-    search: Optional[str] = Query(None),
+    params: Annotated[QueryParams, Query()],
 ):
     """Get all acquisition_cost_data pagination."""
     get_acquisition_cost_data = await get_all(
         db_session=db_session,
-        items_per_page=items_per_page,
-        search=search,
+        items_per_page=params.items_per_page,
+        search=params.search,
         common=common,
     )
     # return

src/manpower_master/schema.py

@@ -3,7 +3,7 @@ from typing import List, Optional
 from uuid import UUID
 
 from pydantic import Field
-from src.models import DefaultBase, Pagination
+from src.models import CommonParams, DefaultBase, Pagination
 
 
 class ManpowerCostBase(DefaultBase):
@@ -31,3 +31,7 @@ class ManpowerCostRead(ManpowerCostBase):
 
 class ManpowerCostPagination(Pagination):
     items: List[ManpowerCostRead] = []
+
+class QueryParams(CommonParams):
+    items_per_page: Optional[int] = Field(5)
+    search: Optional[str] = Field(None)

src/masterdata/router.py

@@ -1,7 +1,9 @@
-from typing import Optional, List
+from typing import Annotated, Optional, List
 from fastapi import APIRouter, HTTPException, status, Query
 
 from sqlalchemy import Select
+
+from src.manpower_cost.schema import QueryParams
 from .model import MasterData
 from .schema import (
     MasterDataPagination,
@@ -23,15 +25,14 @@ router = APIRouter()
 async def get_masterdatas(
     db_session: DbSession,
     common: CommonParameters,
-    items_per_page: Optional[int] = Query(5),
-    search: Optional[str] = Query(None),
+    params: Annotated[QueryParams, Query()],
 ):
     """Get all documents."""
     # return
     master_datas = await get_all(
         db_session=db_session,
-        items_per_page=items_per_page,
-        search=search,
+        items_per_page=params.items_per_page,
+        search=params.search,
         common=common,
     )
     return StandardResponse(

src/masterdata/schema.py

@@ -2,7 +2,7 @@ from datetime import datetime
 from typing import List, Optional
 from uuid import UUID
 
-from pydantic import Field
+from pydantic import BaseModel, Field
 from src.models import DefaultBase, Pagination
 from src.auth.service import CurrentUser
 
@@ -50,3 +50,15 @@ class MasterDataRead(MasterdataBase):
 
 class MasterDataPagination(Pagination):
     items: List[MasterDataRead] = []
+
+
+class QueryParams(BaseModel):
+    items_per_page: Optional[int] = Field(
+        5,
+        ge=1,
+        description="Items per page"
+    )
+    search: Optional[str] = Field(
+        None,
+        description="Search keyword"
+    )

src/masterdata_simulations/router.py

@@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import Annotated, List, Optional
 from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, Query, status
@@ -12,6 +12,7 @@ from src.masterdata_simulations.schema import (
     MasterDataSimulationPagination,
     MasterDataSimulationRead,
     MasterDataSimulationUpdate,
+    QueryParams,
 )
 from src.masterdata_simulations.service import (
     bulk_update,
@@ -30,15 +31,13 @@ router = APIRouter()
 async def get_masterdata_simulations(
     db_session: DbSession,
     common: CommonParameters,
-    simulation_id: UUID = Query(..., description="Simulation identifier"),
-    items_per_page: Optional[int] = Query(5),
-    search: Optional[str] = Query(None),
+    params: Annotated[QueryParams, Query()],
 ):
     master_datas = await get_all(
         db_session=db_session,
-        items_per_page=items_per_page,
-        simulation_id=simulation_id,
-        search=search,
+        items_per_page=params.items_per_page,
+        simulation_id=params.simulation_id,
+        search=params.search,
         common=common,
     )
     return StandardResponse(data=master_datas, message="Data retrieved successfully")

src/masterdata_simulations/schema.py

@@ -4,7 +4,7 @@ from uuid import UUID
 from pydantic import Field
 
 from src.masterdata.schema import MasterdataBase
-from src.models import DefaultBase, Pagination
+from src.models import CommonParams, DefaultBase, Pagination
 
 
 class MasterDataSimulationBase(MasterdataBase):
@@ -38,3 +38,16 @@ class MasterDataSimulationRead(MasterDataSimulationBase):
 
 class MasterDataSimulationPagination(Pagination):
     items: List[MasterDataSimulationRead] = []
+
+
+class QueryParams(CommonParams):
+    simulation_id: UUID = Field(
+        ...,
+        description="Simulation identifier",
+    )
+    items_per_page: Optional[int] = Field(
+        5,
+        ge=1,
+        description="Items per page"
+    )
+    search: Optional[str] = Field(None)

src/models.py

@@ -1,6 +1,6 @@
 # src/common/models.py
 from datetime import datetime
-from typing import Generic, Optional, TypeVar
+from typing import Generic, List, Optional, TypeVar
 import uuid
 from pydantic import BaseModel, Field, SecretStr
 from sqlalchemy import Column, DateTime, String, func, event
@@ -72,6 +72,8 @@ class DefaultBase(BaseModel):
         validate_assignment = True
         arbitrary_types_allowed = True
         str_strip_whitespace = True
+        extra = "forbid"
+        populate_by_name=True
 
         json_encoders = {
             # custom output conversion for datetime
@@ -99,3 +101,21 @@ class StandardResponse(BaseModel, Generic[T]):
     data: Optional[T] = None
     message: str = "Success"
     status: ResponseStatus = ResponseStatus.SUCCESS
+
+
+class CommonParams(DefaultBase):
+    # This ensures no extra query params are allowed
+    current_user: Optional[str] = Field(None, alias="currentUser")
+    page: int = Field(1, gt=0, lt=2147483647)
+    items_per_page: int = Field(5, gt=-2, lt=2147483647)
+    query_str: Optional[str] = Field(None, alias="q")
+    filter_spec: Optional[str] = Field(None, alias="filter")
+    sort_by: List[str] = Field(default_factory=list, alias="sortBy[]")
+    descending: List[bool] = Field(default_factory=list, alias="descending[]")
+    exclude: List[str] = Field(default_factory=list, alias="exclude[]")
+    all_params: int = Field(0, alias="all")
+
+    # Property to mirror your original return dict's bool conversion
+    @property
+    def is_all(self) -> bool:
+        return bool(self.all_params)