Adjusted files for isort & autopep

app/routes/entry.py

@@ -17,6 +17,8 @@ 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
+from app.models.entry import LogType, TransactionType
+from datetime import datetime
 
 router = APIRouter(prefix="/log", tags=["logging"])
 
@@ -62,10 +64,6 @@ 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,
@@ -94,3 +92,37 @@ async def get_logs(
             LogEntry.message.ilike(f"%{search}%") | LogEntry.author.ilike(f"%{search}%")
         )
     return paginate(db, order_by_query(the_select))
+
+
+@router.delete("/", status_code=200)
+async def delete_logs(
+    application: UUID,
+    environment: str | SkipJsonSchema[None] = None,
+    l_type: LogType | SkipJsonSchema[None] = None,
+    t_type: TransactionType | SkipJsonSchema[None] = None,
+    object_reference: str | SkipJsonSchema[None] = None,
+    author: str | SkipJsonSchema[None] = None,
+    sub: str = Security(verify),
+    db: Session = Depends(get_db),
+) -> int:
+    filters = {
+        "application": application,
+        "created_by_id": sub,
+    }
+
+    if environment is not None:
+        filters["environment"] = environment
+    if l_type is not None:
+        filters["l_type"] = l_type
+    if t_type is not None:
+        filters["t_type"] = t_type
+    if object_reference is not None:
+        filters["object_reference"] = object_reference
+    if author is not None:
+        filters["author"] = author
+
+    query = db.query(LogEntry).filter_by(**filters)
+    the_impact = query.count()
+    query.delete(synchronize_session=False)
+    db.commit()
+    return the_impact

app/test_main.py

@@ -239,3 +239,24 @@ class TestAPI:
             re = self.c.get("/log/?application=" + app_id + "&environment=prod")
             assert re["total"] == 2
             assert len(re["results"]) == 2
+
+    def test_logging_delete(self):
+        with log_examples(self) as app_id:
+            re = self.c.delete("/log/?application=" + str(app_id) + "&environment=prod", r_code=200)
+            assert re == 2
+
+            re = self.c.get("/log/?application=" + str(app_id) + "&environment=prod")
+            assert re["total"] == 0
+
+            re = self.c.get("/log/?application=" + str(app_id) + "&environment=dev")
+            assert re["total"] == 3
+
+            # clear complete application
+            re = self.c.get("/log/?application=" + str(app_id))
+            assert re["total"] == 3
+
+            re = self.c.delete("/log/?application=" + str(app_id), r_code=200)
+            assert re == 3
+
+            re = self.c.get("/log/?application=" + str(app_id))
+            assert re["total"] == 0