Allow to filter by tags (#101)

Author: lsbardel

docs/reference/task_plugin.md

@@ -53,6 +53,19 @@ They can be imported from `fluid.scheduler.db`:
 from fluid.scheduler.db import TaskHistoryQuery, TaskRunHistory, TaskRunHistoryPage
 ```
 
+### Filtering by tags
+
+The `tags` field of [TaskHistoryQuery][fluid.scheduler.db.TaskHistoryQuery]
+filters runs by the tags of their [Task][fluid.scheduler.Task]. A run matches
+when its task carries **at least one** of the supplied tags (OR semantics, the
+same as the `tags` query parameter on the task list endpoint). Tags are
+resolved against the live task registry at query time, so they always reflect
+each task's *current* tags rather than the tags it had when the run executed.
+
+When combined with the `task` filter, the two are applied together (AND): the
+run's task must match the name *and* carry one of the tags. Tags that match no
+registered task return an empty result.
+
 ::: fluid.scheduler.db.TaskHistoryQuery
 
 ::: fluid.scheduler.db.TaskRunHistory

docs/release-notes.md

@@ -1,5 +1,13 @@
 # Release Notes
 
+## v2.4.1
+
+Task history can now be filtered by task tags.
+
+- Added a `tags` field to task history queries. Runs match when their task
+  carries at least one of the given tags, resolved against the live registry.
+  ([#101](https://github.com/quantmind/aio-fluid/pull/101))
+
 ## v2.4.0
 
 Lazy settings via pydantic-settings, JSONB params filtering for task

fluid/scheduler/db.py

@@ -80,6 +80,7 @@ def __init__(
 
     def register(self, task_manager: TaskManager) -> None:
         task_manager.state.task_db_plugin = self
+        self.task_manager = task_manager
 
         if is_in_cpu_process():
             return
@@ -140,9 +141,21 @@ async def get_history(
     ) -> TaskRunHistoryPage:
         """Get task run history based on the provided query parameters."""
         table = self.db.tables[self.table_name]
+        filters = q.filters()
+        if q.tags:
+            wanted = set(q.tags)
+            names = {
+                name
+                for name, task in self.task_manager.registry.items()
+                if wanted & task.tags
+            }
+            # AND with an explicit task filter; empty set → IN () → no rows
+            if "name" in filters:
+                names &= {filters["name"]}
+            filters["name"] = list(names)
         pagination = Pagination.create(
             "queued",
-            filters=q.filters(),
+            filters=filters,
             limit=q.limit,
             cursor=q.cursor,
             desc=True,
@@ -289,6 +302,15 @@ class TaskHistoryQuery(BaseModel):
         Query(description="Filter by params using JSON containment"),
         Doc("Filter by params using JSON containment when provided"),
     ] = None
+    tags: Annotated[
+        list[str] | None,
+        Query(description="Filter by task tags (matches any of the given tags)"),
+        Doc(
+            "Filter runs whose task has at least one of these tags when provided. "
+            "Tags are resolved against the live task registry, so they reflect "
+            "each task's current tags."
+        ),
+    ] = None
 
     @model_validator(mode="before")
     @classmethod
@@ -319,7 +341,7 @@ def filters(self) -> dict:
         return {
             self._filter_map.get(k, k): v
             for k, v in self.model_dump(
-                exclude_none=True, exclude={"limit", "cursor"}
+                exclude_none=True, exclude={"limit", "cursor", "tags"}
             ).items()
         }
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "aio-fluid"
-version = "2.4.0"
+version = "2.4.1"
 description = "Tools for backend python services"
 authors = [
     { name = "Luca Sbardella", email = "luca@quantmind.com" },

tests/scheduler/test_db_plugin.py

@@ -304,6 +304,54 @@ async def test_get_history_filter_by_params_http(
     assert all(9.0 == item["params"].get("a") for item in data)
 
 
+async def test_get_history_filter_by_tags(
+    cli_db: TaskClient, task_manager_db: TaskConsumer, db_plugin: TaskDbPlugin
+) -> None:
+    task_run = await task_manager_db.queue_and_wait("dummy", timeout=5)
+    await wait_for_task_run(db_plugin, task_run.id)
+    # "dummy" carries the "slow" tag, "add" has no tags
+    data = await get_history(cli_db, tags=["slow"])
+    assert any(item["id"] == task_run.id for item in data)
+    assert all(item["task"] == "dummy" for item in data)
+
+
+async def test_get_history_filter_by_tags_any(
+    cli_db: TaskClient, task_manager_db: TaskConsumer, db_plugin: TaskDbPlugin
+) -> None:
+    dummy_run = await task_manager_db.queue_and_wait("dummy", timeout=5)
+    fast_run = await task_manager_db.queue_and_wait("fast", timeout=5)
+    await wait_for_task_run(db_plugin, dummy_run.id)
+    await wait_for_task_run(db_plugin, fast_run.id)
+    # OR semantics across tags: "fast" and "dummy" share the "test" tag
+    data = await get_history(cli_db, tags=["fast", "slow"])
+    assert all(item["task"] in {"fast", "dummy"} for item in data)
+    ids = {item["id"] for item in data}
+    assert dummy_run.id in ids
+    assert fast_run.id in ids
+
+
+async def test_get_history_filter_by_tags_no_match(
+    cli_db: TaskClient, task_manager_db: TaskConsumer, db_plugin: TaskDbPlugin
+) -> None:
+    task_run = await task_manager_db.queue_and_wait("dummy", timeout=5)
+    await wait_for_task_run(db_plugin, task_run.id)
+    data = await get_history(cli_db, tags=["does-not-exist"])
+    assert data == []
+
+
+async def test_get_history_filter_by_tags_and_task(
+    cli_db: TaskClient, task_manager_db: TaskConsumer, db_plugin: TaskDbPlugin
+) -> None:
+    task_run = await task_manager_db.queue_and_wait("dummy", timeout=5)
+    await wait_for_task_run(db_plugin, task_run.id)
+    # task + tags combine with AND: "dummy" has "slow" but not "fast"
+    data = await get_history(cli_db, task="dummy", tags=["slow"])
+    assert any(item["id"] == task_run.id for item in data)
+    assert all(item["task"] == "dummy" for item in data)
+    data_empty = await get_history(cli_db, task="dummy", tags=["fast"])
+    assert data_empty == []
+
+
 async def test_get_history_filter_by_params_http_negative(
     cli_db: TaskClient, task_manager_db: TaskConsumer, db_plugin: TaskDbPlugin
 ) -> None: