feat: add range validation constraints to numeric fields across schemas and services

oh_security
Cizz22 1 month ago
parent 20ea1b580a
commit 1029cd5ca9

@ -33,7 +33,7 @@ class OverhaulRead(OverhaulBase):
class BudgetContraintQuery(DefultBase): class BudgetContraintQuery(DefultBase):
cost_threshold: float = 100 cost_threshold: float = Field(100, ge=0, le=1000000000000000)
# { # {
# "overview": { # "overview": {

@ -7,14 +7,14 @@ from src.models import DefultBase
class CommonParams(DefultBase): class CommonParams(DefultBase):
# This ensures no extra query params are allowed # This ensures no extra query params are allowed
current_user: Optional[str] = Field(None, alias="currentUser") current_user: Optional[str] = Field(None, alias="currentUser")
page: int = Field(1, gt=0, lt=2147483647) page: int = Field(1, gt=0, le=1000000)
items_per_page: int = Field(5, gt=0, le=50, multiple_of=5, alias="itemsPerPage") items_per_page: int = Field(5, gt=0, le=50, multiple_of=5, alias="itemsPerPage")
query_str: Optional[str] = Field(None, alias="q") query_str: Optional[str] = Field(None, alias="q")
filter_spec: Optional[str] = Field(None, alias="filter") filter_spec: Optional[str] = Field(None, alias="filter")
sort_by: List[str] = Field(default_factory=list, alias="sortBy[]") sort_by: List[str] = Field(default_factory=list, alias="sortBy[]")
descending: List[bool] = Field(default_factory=list, alias="descending[]") descending: List[bool] = Field(default_factory=list, alias="descending[]")
exclude: List[str] = Field(default_factory=list, alias="exclude[]") exclude: List[str] = Field(default_factory=list, alias="exclude[]")
all_params: int = Field(0, alias="all") all_params: int = Field(0, alias="all", ge=0, le=1000000)
# Property to mirror your original return dict's bool conversion # Property to mirror your original return dict's bool conversion
@property @property

@ -20,14 +20,14 @@ QueryStr = constr(pattern=r"^[ -~]+$", min_length=1)
def common_parameters( def common_parameters(
db_session: DbSession, # type: ignore db_session: DbSession, # type: ignore
current_user: QueryStr = Query(None, alias="currentUser"), # type: ignore current_user: QueryStr = Query(None, alias="currentUser"), # type: ignore
page: int = Query(1, gt=0, lt=2147483647), page: int = Query(1, gt=0, le=1000000),
items_per_page: int = Query(5, alias="itemsPerPage", gt=-2, lt=2147483647), items_per_page: int = Query(5, alias="itemsPerPage", gt=-2, le=1000000),
query_str: QueryStr = Query(None, alias="q"), # type: ignore query_str: QueryStr = Query(None, alias="q"), # type: ignore
filter_spec: QueryStr = Query(None, alias="filter"), # type: ignore filter_spec: QueryStr = Query(None, alias="filter"), # type: ignore
sort_by: List[str] = Query([], alias="sortBy[]"), sort_by: List[str] = Query([], alias="sortBy[]"),
descending: List[bool] = Query([], alias="descending[]"), descending: List[bool] = Query([], alias="descending[]"),
exclude: List[str] = Query([], alias="exclude[]"), exclude: List[str] = Query([], alias="exclude[]"),
all: int = Query(0), all: int = Query(0, ge=0, le=1000000),
# role: QueryStr = Depends(get_current_role), # role: QueryStr = Depends(get_current_role),
): ):
return { return {

@ -13,7 +13,7 @@ class ScopeEquipmentActivityBase(DefultBase):
class ScopeEquipmentActivityCreate(ScopeEquipmentActivityBase): class ScopeEquipmentActivityCreate(ScopeEquipmentActivityBase):
name: str name: str
cost: Optional[float] = Field(0) cost: Optional[float] = Field(0, ge=0, le=1000000000000000)
class ScopeEquipmentActivityUpdate(ScopeEquipmentActivityBase): class ScopeEquipmentActivityUpdate(ScopeEquipmentActivityBase):
@ -23,7 +23,7 @@ class ScopeEquipmentActivityUpdate(ScopeEquipmentActivityBase):
class ScopeEquipmentActivityRead(ScopeEquipmentActivityBase): class ScopeEquipmentActivityRead(ScopeEquipmentActivityBase):
name: str name: str
cost: float cost: float = Field(..., ge=0, le=1000000000000000)
class ScopeEquipmentActivityPagination(Pagination): class ScopeEquipmentActivityPagination(Pagination):

@ -17,14 +17,14 @@ class OverhaulActivityCreate(OverhaulActivityBase):
class OverhaulActivityUpdate(OverhaulActivityBase): class OverhaulActivityUpdate(OverhaulActivityBase):
material_cost: Optional[float] = Field(0) material_cost: Optional[float] = Field(0, ge=0, le=1000000000000000)
service_cost: Optional[float] = Field(0) service_cost: Optional[float] = Field(0, ge=0, le=1000000000000000)
class OverhaulScope(DefultBase): class OverhaulScope(DefultBase):
type: str type: str
start_date: datetime start_date: datetime
end_date: datetime end_date: datetime
duration_oh: int duration_oh: int = Field(..., ge=0, le=1000000)
class ScopeEquipmentJob(DefultBase): class ScopeEquipmentJob(DefultBase):
@ -35,12 +35,12 @@ class OverhaulJob(DefultBase):
class OverhaulActivityRead(OverhaulActivityBase): class OverhaulActivityRead(OverhaulActivityBase):
id: UUID id: UUID
material_cost: Optional[float] = Field(0) material_cost: Optional[float] = Field(0, ge=0, le=1000000000000000)
service_cost: Optional[float] = Field(0) service_cost: Optional[float] = Field(0, ge=0, le=1000000000000000)
location_tag: str location_tag: str
equipment_name: Optional[str] equipment_name: Optional[str]
oh_scope: str oh_scope: str
overhaul_cost: Optional[float] = Field(0) overhaul_cost: Optional[float] = Field(0, ge=0, le=1000000000000000)
# equipment: MasterEquipmentTree # equipment: MasterEquipmentTree
# overhaul_scope: OverhaulScope # overhaul_scope: OverhaulScope
# overhaul_jobs: Optional[List[OverhaulJob]] = Field([]) # overhaul_jobs: Optional[List[OverhaulJob]] = Field([])

@ -8,8 +8,8 @@ from src.models import DefultBase, Pagination
class ScopeBase(DefultBase): class ScopeBase(DefultBase):
duration_oh: Optional[int] = Field(720, title="Duration OH") duration_oh: Optional[int] = Field(720, title="Duration OH", ge=0, le=1000000)
crew_number: Optional[int] = Field(10, title="Crew") crew_number: Optional[int] = Field(10, title="Crew", ge=0, le=1000000)
status: Optional[str] = Field("Upcoming") status: Optional[str] = Field("Upcoming")

Loading…
Cancel
Save