API: Add an endpoint to check the history of a domain on a list

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

diff --git a/src/dnsbl/api/lists.py b/src/dnsbl/api/lists.py
index 766431402c7b5777ce93416fe0a2f9ae9850264a..8085ae511edde92dc80a2f1935b4553a96359939 100644 (file)
--- a/src/dnsbl/api/lists.py
+++ b/src/dnsbl/api/lists.py
@@ -83,6 +83,38 @@ def get_list_reports(
 ) -> typing.List[reports.Report]:
        return list.get_reports(open=open, limit=limit)
 
+@router.get("/{list}/domains/{name}")
+def get_list_domains(
+               name: str, list = fastapi.Depends(get_list_from_path),
+):
+       # Fetch the domain history
+       events = list.get_domain_history(name)
+
+       # Return all events
+       for event in events:
+               e = {
+                       "timestamp" : event.ts,
+                       "type"      : event.type,
+                       "block"     : event.block,
+               }
+
+               # Append the person who has done thiy
+               if event.by:
+                       e["by"] = event.by
+
+               # Append the source
+               if event.source_id:
+                       source = backend.sources.get_by_id(event.source_id)
+
+                       if source:
+                               e["source"] = source.name
+
+               # Append the report
+               if event.report:
+                       e["report"] = event.report
+
+               yield e
+
 class CreateReport(pydantic.BaseModel):
        # Domain
        name : str

diff --git a/src/dnsbl/lists.py b/src/dnsbl/lists.py
index 60b6e8603f2883b17fe95b80cf613f6f0a4fd6b5..1eef127a0eb7d89a230fe680466fff3bbe009c89 100644 (file)
--- a/src/dnsbl/lists.py
+++ b/src/dnsbl/lists.py
@@ -736,6 +736,68 @@ class List(sqlmodel.SQLModel, database.BackendMixin, table=True):
 
                return self.backend.db.select(stmt)
 
+       def get_domain_history(self, name):
+               """
+                       Fetches the history the given domain
+               """
+               events = (
+                       sqlmodel.union_all(
+                               # Fetch all events where a domain has been added
+                               sqlmodel
+                               .select(
+                                       domains.Domain.name.label("name"),
+                                       domains.Domain.added_at.label("ts"),
+                                       sqlmodel.literal("added").label("type"),
+                                       domains.Domain.block.label("block"),
+                                       domains.Domain.added_by.label("by"),
+                                       domains.Domain.source_id.label("source_id"),
+                                       domains.Domain.report_add_id.label("report"),
+                                       sqlmodel.literal(1).label("delta"),
+                               ).
+                               where(
+                                       domains.Domain.list == self,
+                                       domains.Domain.name == name,
+                               ),
+
+                               # Fetch all events where a domain has been removed
+                               sqlmodel
+                               .select(
+                                       domains.Domain.name.label("name"),
+                                       domains.Domain.removed_at.label("ts"),
+                                       sqlmodel.literal("removed").label("type"),
+                                       domains.Domain.block.label("block"),
+                                       domains.Domain.removed_by.label("by"),
+                                       domains.Domain.source_id.label("source_id"),
+                                       domains.Domain.report_remove_id.label("report"),
+                                       sqlmodel.literal(-1).label("delta"),
+                               )
+                               .where(
+                                       domains.Domain.list == self,
+                                       domains.Domain.name == name,
+                                       domains.Domain.removed_at != None,
+                               )
+                       )
+                       .cte("events")
+               )
+
+               # Order all events chronologically
+               stmt = (
+                       sqlmodel
+                       .select(
+                               events.c.ts,
+                               events.c.type,
+                               events.c.block,
+                               events.c.by,
+                               events.c.source_id,
+                               events.c.report,
+                       )
+                       .order_by(
+                               events.c.ts,
+                       )
+               )
+
+               return self.backend.db.select(stmt)
+
 
 class ListStats(sqlmodel.SQLModel, table=True):
        __tablename__ = "list_stats"