feat: add delete operation

Author: martynia

diracx-db/src/diracx/db/os/pilot_logs.py

@@ -7,6 +7,7 @@ class PilotLogsDB(BaseOSDB):
     fields = {
         "PilotStamp": {"type": "keyword"},
         "PilotID": {"type": "long"},
+        "SubmissionTime": {"type": "date"},
         "LineNumber": {"type": "long"},
         "Message": {"type": "text"},
         "VO": {"type": "keyword"},

diracx-routers/src/diracx/routers/pilot_logging/access_policies.py

@@ -39,7 +39,7 @@ async def policy(
         pilot_db: PilotLogsDB | None = None,
         pilot_ids: list[int] | None = None,  # or pilot stamp list ?
     ):
-        print("user_info.properties:", user_info.properties)
+
         assert action, "action is a mandatory parameter"
         assert pilot_db, "pilot_db is a mandatory parameter"
 

diracx-routers/src/diracx/routers/pilot_logging/remote_logger.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import datetime
+
 from pydantic import BaseModel
 from sqlalchemy import select
 
+from diracx.core.exceptions import InvalidQueryError
 from diracx.db.sql.pilot_agents.schema import PilotAgents
 from diracx.db.sql.utils import BaseSQLDB
 
@@ -25,6 +28,12 @@ class LogMessage(BaseModel):
     vo: str
 
 
+class DateRange(BaseModel):
+    pilot_id: int | None = None
+    min: str | None = None  # expects a string in ISO 8601 ("%Y-%m-%dT%H:%M:%S.%f%z")
+    max: str | None = None  # expects a string in ISO 8601 ("%Y-%m-%dT%H:%M:%S.%f%z")
+
+
 @router.post("/")
 async def send_message(
     data: LogMessage,
@@ -35,23 +44,27 @@ async def send_message(
     await check_permissions(action=ActionType.CREATE, pilot_db=pilot_logs_db)
 
     pilot_id = 0  # need to get pilot id from pilot_stamp (via PilotAgentsDB)
+    # also add a timestamp to be able to select and delete logs based on pilot creation dates, even if corresponding
+    # pilots have been already deleted from PilotAgentsDB (so the logs can live longer than pilots).
+    submission_time = datetime.datetime.fromtimestamp(0, datetime.timezone.utc)
     piloAgentsDB = BaseSQLDB.available_implementations("PilotAgentsDB")[0]
     url = BaseSQLDB.available_urls()["PilotAgentsDB"]
     db = piloAgentsDB(url)
 
     async with db.engine_context():
         async with db:
-            stmt = select(PilotAgents.PilotID).where(
+            stmt = select(PilotAgents.PilotID, PilotAgents.SubmissionTime).where(
                 PilotAgents.PilotStamp == data.pilot_stamp
             )
-            pilot_id = (await db.conn.execute(stmt)).scalar_one()
+            pilot_id, submission_time = (await db.conn.execute(stmt)).one()
 
     docs = []
     for line in data.lines:
         docs.append(
             {
                 "PilotStamp": data.pilot_stamp,
                 "PilotID": pilot_id,
+                "SubmissionTime": submission_time,
                 "VO": data.vo,
                 "LineNumber": line.line_no,
                 "Message": line.line,
@@ -72,21 +85,35 @@ async def get_logs(
 
     result = await db.search(
         ["Message"],
-        [{"parameter": "PilotID", "operator": "eq"} | {"value": pilot_id}],
+        [{"parameter": "PilotID", "operator": "eq", "value": pilot_id}],
         [{"parameter": "LineNumber", "direction": "asc"}],
     )
     if not result:
         return [f"No logs for pilot ID = {pilot_id}"]
     return result
 
 
-@router.delete("/delete")
-async def delete_by_pilot_id(
-    pilot_id: int,
+@router.delete("/logs")
+async def delete(
+    data: DateRange,
     db: PilotLogsDB,
     check_permissions: CheckPilotLogsPolicyCallable,
-):
-    logger.warning(f"Deleting logs for pilot ID '{pilot_id}'")
+) -> str:
+    """Delete either logs for a specific PilotID or a creation date range."""
     await check_permissions(action=ActionType.DELETE, pilot_db=db)
-    await db.delete([{"parameter": "PilotID", "operator": "eq"} | {"value": pilot_id}])
-    return f"Logs for pilot ID '{pilot_id}' successfully deleted"
+    if data.pilot_id:
+        await db.delete(
+            [{"parameter": "PilotID", "operator": "eq", "value": data.pilot_id}]
+        )
+        return f"Logs for pilot ID '{data.pilot_id}' successfully deleted"
+    if data.min and not data.max:
+        logger.warning(f"Deleting logs for pilots with submission data >='{data.min}'")
+        await db.delete(
+            [{"parameter": "SubmissionTime", "operator": "gt", "value": data.min}]
+        )
+        return f"Logs for for pilots with submission data >='{data.min}' successfully deleted"
+    if data.min and data.max:
+        raise InvalidQueryError(
+            "This query requires a range operater definition in DiracX"
+        )
+    return "no-op"