feat: added filters for logs

app/routes/entry.py

@@ -2,7 +2,7 @@ from creyPY.fastapi.crud import (
     create_obj_from_data,
 )
 from creyPY.fastapi.order_by import order_by
-from typing import Any, Callable
+from typing import Callable
 from sqlalchemy.sql.selectable import Select
 from creyPY.fastapi.db.session import get_db
 from fastapi import APIRouter, Depends, Security, HTTPException
@@ -15,6 +15,8 @@ from fastapi_pagination.ext.sqlalchemy import paginate
 from creyPY.fastapi.pagination import Page
 from uuid import UUID
 from pydantic.json_schema import SkipJsonSchema
+from fastapi_filters import FilterValues, create_filters
+from fastapi_filters.ext.sqlalchemy import apply_filters
 
 router = APIRouter(prefix="/log", tags=["logging"])
 
@@ -60,19 +62,32 @@ async def get_log(
     return LogOUT.model_validate(obj)
 
 
+from app.models.entry import LogType, TransactionType
+from datetime import datetime
+
+
 @router.get("/")
 async def get_logs(
     search: str | SkipJsonSchema[None] = None,
     order_by_query: Callable[[Select], Select] = Depends(order_by),
+    filters: FilterValues = Depends(
+        create_filters(
+            created_by_id=str,
+            l_type=LogType,
+            t_type=TransactionType,
+            application=UUID,
+            object_reference=str,
+            author=str,
+            created_at=datetime,
+        )
+    ),
     sub: str = Security(verify),
     db: Session = Depends(get_db),
 ) -> Page[LogOUT]:
     """
     Filter logs of your systems. Searching works only for author and message. Use filters for the rest.
     """
-    # add filters
+    the_select = apply_filters(select(LogEntry).filter(LogEntry.created_by_id == sub), filters)
-    # add sorting
-    the_select = select(LogEntry).filter(LogEntry.created_by_id == sub)
     if search:
         the_select = the_select.filter(
             LogEntry.message.ilike(f"%{search}%") | LogEntry.author.ilike(f"%{search}%")

app/test_main.py

@@ -41,7 +41,7 @@ def log_examples(self):
             "object_reference": "1",
             "previous_object": {"name": "Unit 1"},
         },
-        {"l_type": "info", "t_type": "delete", "message": "User Max Mustermann deleted"},
+        {"l_type": "warning", "t_type": "delete", "message": "User Max Mustermann deleted"},
     ]
     with app_context(self) as app_id:
         for entry in LOG_EXAMPLES:
@@ -165,5 +165,49 @@ class TestAPI:
             assert len(re["results"]) == 5
             assert re["results"][0]["created_at"] > re["results"][1]["created_at"]
 
-    # def test_logging_filter(self):
+    def test_logging_filter(self):
-    #     pass
+        with log_examples(self) as app_id:
+            # API KEY
+            re = self.c.get("/log/?created_by_id=" + CURRENT_USER)
+            assert re["total"] == 5
+            assert len(re["results"]) == 5
+
+            # LogType
+            re = self.c.get("/log/?l_type=info")
+            assert re["total"] == 4
+            assert len(re["results"]) == 4
+
+            # TransactionType
+            re = self.c.get("/log/?t_type=create")
+            assert re["total"] == 2
+            assert len(re["results"]) == 2
+
+            # TransactipnType create and update
+            re = self.c.get("/log/?t_type%5Bin%5D=create,update")
+            assert re["total"] == 4
+            assert len(re["results"]) == 4
+
+            # Application
+            re = self.c.get("/log/?application=" + app_id)
+            assert re["total"] == 5
+            assert len(re["results"]) == 5
+
+            # Application not
+            re = self.c.get("/log/?application%5Bne%5D=" + app_id)
+            assert re["total"] == 0
+            assert len(re["results"]) == 0
+
+            # Object Reference
+            re = self.c.get("/log/?object_reference=1")
+            assert re["total"] == 2
+            assert len(re["results"]) == 2
+
+            # author
+            re = self.c.get("/log/?author=auth|max_muster")
+            assert re["total"] == 2
+            assert len(re["results"]) == 2
+
+            # not author
+            re = self.c.get("/log/?author%5Bne%5D=auth|max_muster")
+            assert re["total"] == 3
+            assert len(re["results"]) == 3

requirements.txt

@@ -32,3 +32,5 @@ iniconfig==2.0.0 # pytest
 packaging==24.1 # pytest
 pluggy==1.5.0 # pytest
 pytest==8.3.3 # pytest
+
+fastapi-filters==0.2.9 # Filters