lists: Implement fetching reports over the API

Signed-off-by: Michael Tremer <michael.tremer@ipfire.org>

diff --git a/src/dnsbl/api/lists.py b/src/dnsbl/api/lists.py
index c2bf310bd8c66e56322356ce25b8de015cabbc45..069c2ce3b453a63e9430cbf4aebaf9862abfb29b 100644 (file)
--- a/src/dnsbl/api/lists.py
+++ b/src/dnsbl/api/lists.py
@@ -58,6 +58,14 @@ def get_list(list = fastapi.Depends(get_list_from_path)) -> lists.List:
 def get_list_sources(list = fastapi.Depends(get_list_from_path)) -> typing.List[sources.Source]:
        return list.sources
 
+@router.get("/{list}/reports")
+def get_list_reports(
+               list = fastapi.Depends(get_list_from_path),
+               open: bool | None = None,
+               limit: int | None = None
+) -> typing.List[reports.Report]:
+       return list.get_reports(open=open, limit=limit)
+
 class CreateReport(pydantic.BaseModel):
        # Domain
        name : str

diff --git a/src/dnsbl/lists.py b/src/dnsbl/lists.py
index 276f6893e63055889ff30d376dbfff2aeea0d18f..125ce02e97b11dc70126890e833194462a236d16 100644 (file)
--- a/src/dnsbl/lists.py
+++ b/src/dnsbl/lists.py
@@ -28,6 +28,7 @@ import typing
 from . import checker
 from . import database
 from . import exporters
+from . import reports
 from . import sources
 from . import util
 from .i18n import _
@@ -331,6 +332,39 @@ class List(sqlmodel.SQLModel, database.BackendMixin, table=True):
        # Reports
        reports : typing.List["Report"] = sqlmodel.Relationship(back_populates="list")
 
+       def get_reports(self, open=None, limit=None):
+               """
+                       Fetches the most recent reports
+               """
+               stmt = (
+                       sqlmodel
+                       .select(
+                               reports.Report,
+                       )
+                       .where(
+                               reports.Report.list == self,
+                       )
+                       .order_by(
+                               reports.Report.reported_at.desc(),
+                       )
+               )
+
+               # Filter for open/closed reports only
+               if open is True:
+                       stmt = stmt.where(
+                               reports.Report.closed_at == False,
+                       )
+               elif open is False:
+                       stmt = stmt.where(
+                               reports.Report.closed_at == True,
+                       )
+
+               # Optionally apply the limit
+               if limit:
+                       stmt = stmt.limit(limit)
+
+               return self.backend.db.fetch_as_list(stmt)
+
        # Report!
 
        def report(self, **kwargs):