Merge pull request #4 from guivaloz:guivaloz/personas-param

Personas param

perseo/v4/personas/crud.py

@@ -6,7 +6,7 @@
 from sqlalchemy.orm import Session
 
 from lib.exceptions import MyIsDeletedError, MyNotExistsError, MyNotValidParamError
-from lib.safe_string import safe_rfc
+from lib.safe_string import safe_curp, safe_rfc, safe_string
 from perseo.v4.tabuladores.crud import get_tabulador
 
 from ...core.personas.models import Persona
@@ -15,12 +15,41 @@
 def get_personas(
     database: Session,
     tabulador_id: int = None,
+    rfc: str = None,
+    nombres: str = None,
+    apellido_primero: str = None,
+    apellido_segundo: str = None,
+    curp: str = None,
 ) -> Any:
     """Consultar las personas activos"""
     consulta = database.query(Persona)
     if tabulador_id is not None:
         tabulador = get_tabulador(database, tabulador_id)
         consulta = consulta.filter_by(tabulador_id=tabulador.id)
+    if rfc is not None:
+        try:
+            rfc = safe_rfc(rfc, search_fragment=True)
+        except ValueError as error:
+            raise MyNotValidParamError(str(error)) from error
+        consulta = consulta.filter(Persona.rfc.contains(rfc))
+    if nombres is not None:
+        nombres = safe_string(nombres, save_enie=True)
+        if nombres != "":
+            consulta = consulta.filter(Persona.nombres.contains(nombres))
+    if apellido_primero is not None:
+        apellido_primero = safe_string(apellido_primero, save_enie=True)
+        if apellido_primero != "":
+            consulta = consulta.filter(Persona.apellido_primero.contains(apellido_primero))
+    if apellido_segundo is not None:
+        apellido_segundo = safe_string(apellido_segundo, save_enie=True)
+        if apellido_segundo != "":
+            consulta = consulta.filter(Persona.apellido_segundo.contains(apellido_segundo))
+    if curp is not None:
+        try:
+            curp = safe_curp(curp, search_fragment=True)
+        except ValueError as error:
+            raise MyNotValidParamError(str(error)) from error
+        consulta = consulta.filter(Persona.curp.contains(curp))
     return consulta.filter_by(estatus="A").order_by(Persona.id)
 
 

perseo/v4/personas/paths.py

@@ -23,6 +23,11 @@ async def paginado_personas(
     current_user: Annotated[UsuarioInDB, Depends(get_current_active_user)],
     database: Annotated[Session, Depends(get_db)],
     tabulador_id: int = None,
+    rfc: str = None,
+    nombres: str = None,
+    apellido_primero: str = None,
+    apellido_segundo: str = None,
+    curp: str = None,
 ):
     """Paginado de personas"""
     if current_user.permissions.get("PERSONAS", 0) < Permiso.VER:
@@ -31,6 +36,11 @@ async def paginado_personas(
         resultados = get_personas(
             database=database,
             tabulador_id=tabulador_id,
+            rfc=rfc,
+            nombres=nombres,
+            apellido_primero=apellido_primero,
+            apellido_segundo=apellido_segundo,
+            curp=curp,
         )
     except MyAnyError as error:
         return CustomPage(success=False, message=str(error))