svcs|routes.turnilo_dashboards: adopt Query class

Adopt Oscar's suggestion to use a Query class to validate query
parameters.

Unfortunately, due to the circular dependency, the Query class
is moved to services (due to typing).

Signed-off-by: Marc Sune <[email protected]>
Signed-off-by: Oscar Moya <[email protected]>

src/routes/turnilo_dashboard_routes.py

@@ -1,5 +1,4 @@
-import re
-from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import APIRouter, Depends, HTTPException
 from typing import List
 from sqlmodel import Session
 from models.turnilo_dashboard import TurniloDashboard
@@ -13,27 +12,15 @@
 ### Dashboards ###
 
 
-def is_valid_query(s: str) -> bool:
-    if not isinstance(s, str):
-        return False
-    if len(s) > 256:
-        return False
-    pattern = r'^[a-zA-Z0-9_-]+$'
-    return bool(re.match(pattern, s))
-
-
 @turnilo_router.get(
     constants.URL_PATH + "/turnilo/dashboards/",
     response_model=List[TurniloDashboard],
     summary="Gets all Turnilo dashboards"
 )
 def turnilo_get_dashboards(db_session: Session = Depends(db.get_session),
-                           shortName: str = Query(None), dataCube: str = Query(None)):
-    if shortName and not is_valid_query(shortName):
-        raise HTTPException(status_code=400, detail=f"Invalid shortName='{shortName}'")
-    if dataCube and not is_valid_query(dataCube):
-        raise HTTPException(status_code=400, detail=f"Invalid dataCube='{dataCube}'")
-    return td.dashboards_get_all(db_session, shortName, dataCube)
+                           query_params: td.GetQueryParams = Depends()):
+    query_params.validate()
+    return td.dashboards_get_all(db_session, query_params)
 
 
 @turnilo_router.get(

src/services/turnilo_dashboards.py

@@ -1,18 +1,40 @@
-from typing import List
+import re
+from typing import List, Optional
 from sqlmodel import Session, select
+from pydantic import BaseModel, Field
 from sqlalchemy import exc
 from models.turnilo_dashboard import TurniloDashboard
 from fastapi import HTTPException
 
 # Turnilo Dashboards
 
 
-def dashboards_get_all(session: Session, shortName: str, dataCube: str) -> List[TurniloDashboard]:
+class GetQueryParams(BaseModel):
+    """
+    Get query filtering params
+    """
+    shortName: Optional[str] = Field(default=None, description="Dashboard's shortName")
+    dataCube: Optional[str] = Field(default=None, description="Dashboard's dataCube")
+
+    def is_valid_param(self, s: str) -> bool:
+        if len(s) > 256:
+            return False
+        pattern = r'^[a-zA-Z0-9_-]+$'
+        return bool(re.match(pattern, s))
+
+    def validate(self):
+        if self.shortName and not self.is_valid_param(self.shortName):
+            raise HTTPException(status_code=400, detail=f"Invalid shortName='{self.shortName}'")
+        if self.dataCube and not self.is_valid_param(self.dataCube):
+            raise HTTPException(status_code=400, detail=f"Invalid dataCube='{self.dataCube}'")
+
+
+def dashboards_get_all(session: Session, query_params: GetQueryParams) -> List[TurniloDashboard]:
     statement = select(TurniloDashboard)
-    if shortName:
-        statement = statement.where(TurniloDashboard.shortName == shortName)
-    if dataCube:
-        statement = statement.where(TurniloDashboard.dataCube == dataCube)
+    if query_params.shortName:
+        statement = statement.where(TurniloDashboard.shortName == query_params.shortName)
+    if query_params.dataCube:
+        statement = statement.where(TurniloDashboard.dataCube == query_params.dataCube)
     return list(session.exec(statement).all())