Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -132,11 +132,12 @@ def find_n_latest_nodes_ids(
(_validate_date_range, (min_node_date, max_node_date)),
]
node_filters = [(_validate_node_id, (node_id_min_val, node_id_max_val))]
name_pattern = (
f"#*{search_filter.name_part}*"
if search_filter is not None and search_filter.name_part is not None
else "#*"
)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should probably change if design to if search_filter : and than in the nested ifs do the other ones, also would be nice to have comments describing the synax since f"#{search_filter.name}" is unclear

if search_filter is not None and search_filter.name is not None:
name_pattern = f"#*_{search_filter.name}_*"
elif search_filter is not None and search_filter.name_part is not None:
name_pattern = f"#*{search_filter.name_part}*"
else:
name_pattern = "#*"
next_ = None
for node_date in sorted(
filter(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,9 @@ def _fill_date(self, dt: date | None = None) -> None:
for node_path in to_add:
self._add_node(node_path)

def _get_suited_ids_by_name(self, name: str) -> set[IdType]:
return set(self._name2id.get(name, []))

def _get_suited_ids_by_name_part(self, name_part: str) -> set[IdType]:
return set(
chain.from_iterable(
Expand Down Expand Up @@ -130,7 +133,9 @@ def get_ids(
):
return set()
allowed_ids: set[IdType] | None = None
if filters.name_part:
if filters.name:
allowed_ids = self._get_suited_ids_by_name(filters.name)
elif filters.name_part:
allowed_ids = self._get_suited_ids_by_name_part(filters.name_part)
if allowed_ids is not None and len(allowed_ids) == 0:
return set()
Expand Down
24 changes: 19 additions & 5 deletions backend/qualibrate_app/api/core/domain/timeline_db/root.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,13 +33,27 @@ def _get_latest_snapshots(
reverse: bool,
) -> tuple[int, DocumentSequenceType]:
timeline_db_config = self.timeline_db_config
params: dict[str, Any] = {
"page": pages_filter.page,
"per_page": pages_filter.per_page,
"reverse": reverse,
}
if search_filter is not None:

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

instead of putting if for every parameter, we can iterate through parameters and assign if not value

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

also, do we use timelinedb? do we use the added parameters?

if search_filter.name is not None:
params["name"] = search_filter.name
if search_filter.name_part is not None:
params["name_part"] = search_filter.name_part
if search_filter.min_node_id != 1:
params["min_node_id"] = search_filter.min_node_id
if search_filter.max_node_id is not None:
params["max_node_id"] = search_filter.max_node_id
if search_filter.min_date is not None:
params["min_date"] = search_filter.min_date.isoformat()
if search_filter.max_date is not None:
params["max_date"] = search_filter.max_date.isoformat()
result = request_with_db(
"snapshot/n_latest",
params={
"page": pages_filter.page,
"per_page": pages_filter.per_page,
"reverse": reverse,
},
params=params,
db_name=self._settings.project,
host=timeline_db_config.address_with_root,
timeout=timeline_db_config.timeout,
Expand Down
22 changes: 22 additions & 0 deletions backend/qualibrate_app/api/core/types.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,33 @@
import string
from collections.abc import Mapping, Sequence
from datetime import date
from enum import Enum
from typing import Any

from pydantic import BaseModel

AllowedSearchKeys = string.ascii_letters + string.digits + "-_*"


Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

great mechanism! can you leave some comments explaining it for future developers?

class SortField(str, Enum):
"""Supported fields for sorting snapshots."""

name = "name"
date = "date"
status = "status"


# Priority order for status sorting (lower number = higher priority)
# Used when sorting with descending=False (ascending)
STATUS_SORT_PRIORITY: dict[str | None, int] = {
"finished": 0,
"skipped": 1,
"pending": 2,
"running": 3,
"error": 4,
None: 5,
}

IdType = int
DocumentType = Mapping[str, Any]
DocumentSequenceType = Sequence[DocumentType]
Expand All @@ -18,6 +39,7 @@ class PageFilter(BaseModel):


class SearchFilter(BaseModel):
name: str | None = None
name_part: str | None = None
min_node_id: IdType = 1
max_node_id: int | None = None
Expand Down
2 changes: 1 addition & 1 deletion backend/qualibrate_app/api/core/utils/find_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ def _get_subpath_value_wildcard(
else:
return []

if not isinstance(obj, (Mapping, Sequence)):

@Elad-Zaharan Elad-Zaharan Jan 22, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why is this change needed? it seems less conventional, am i missing something?

if not isinstance(obj, Mapping | Sequence):
return []
iter_function = cast(
Callable[..., Sequence[tuple[str, Any]] | Sequence[tuple[int, Any]]],
Expand Down
96 changes: 94 additions & 2 deletions backend/qualibrate_app/api/routes/root.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from collections.abc import Sequence
from typing import Annotated, Any

from fastapi import APIRouter, Depends, Query
from fastapi import APIRouter, Depends, HTTPException, Query
from qualibrate_config.models import QualibrateConfig, StorageType

from qualibrate_app.api.core.domain.bases.branch import BranchLoadType
Expand All @@ -21,10 +21,12 @@
)
from qualibrate_app.api.core.models.snapshot import Snapshot as SnapshotModel
from qualibrate_app.api.core.types import (
STATUS_SORT_PRIORITY,
IdType,
PageFilter,
SearchFilter,
SearchWithIdFilter,
SortField,
)
from qualibrate_app.api.dependencies.search import get_search_path
from qualibrate_app.api.routes.utils.dependencies import (
Expand Down Expand Up @@ -367,6 +369,29 @@ def get_snapshot_filtered(
return snapshots[0].dump()


def _get_sort_key(
snapshot: SimplifiedSnapshotWithMetadata,
sort_field: SortField,
descending: bool,
) -> tuple[int, Any]:
"""Get sort key for a snapshot based on the sort field.

Returns a tuple of (priority, value) for stable sorting.
"""
if sort_field == SortField.name:
name = snapshot.metadata.name if snapshot.metadata else None
# Use empty string for None to sort nulls last
return (0 if name else 1, (name or "").lower())
elif sort_field == SortField.date:
# Use created_at for date sorting
return (0 if snapshot.created_at else 1, snapshot.created_at)
elif sort_field == SortField.status:
status = snapshot.metadata.status if snapshot.metadata else None
priority = STATUS_SORT_PRIORITY.get(status, len(STATUS_SORT_PRIORITY))
return (priority, status or "")
return (0, snapshot.id)


@root_router.get("/snapshots_history", summary="List snapshots history")
def get_snapshots_history(
*,
Expand All @@ -389,14 +414,28 @@ def get_snapshots_history(
description="This field is ignored. Use `descending` instead.",
),
] = False,
sort: Annotated[
SortField | None,
Query(
description=(
"Field to sort by: 'name' (alphabetical), 'date' (creation "
"time), or 'status' (finished first, then skipped, pending, "
"running, error). Default sorts by date."
)
),
] = None,
search_filters: Annotated[SearchFilter, Depends(get_search_filter)],
root: Annotated[RootBase, Depends(_get_root_instance)],
) -> PagedCollection[SimplifiedSnapshotWithMetadata]:
"""List snapshots history.

Returns a paginated list of snapshots for the storage. Order is controlled
by the `descending` flag.
by the `descending` flag. Optionally filter by date range, name, or node ID.
Use the `sort` parameter to sort by name, date, or status.

### Examples

#### 1) Basic pagination
**Request:** `/root/snapshots_history?page=1&per_page=3`

**Response:**
Expand Down Expand Up @@ -445,15 +484,68 @@ def get_snapshots_history(
"total_pages": 45
}
```

#### 2) Filter by date range
**Request:**
`/root/snapshots_history?page=1&per_page=100&min_date=2025-08-21&max_date=2025-08-22`

Returns snapshots created between 2025-08-21 and 2025-08-22 (inclusive).

#### 3) Filter by name substring and date
**Request:**
`/root/snapshots_history?page=1&per_page=50&name_part=test_cal&min_date=2025-08-01`

Returns snapshots with name containing "test_cal" created on or after
2025-08-01.

#### 4) Filter by exact name
**Request:**
`/root/snapshots_history?page=1&per_page=100&name=test_cal`

Returns only snapshots with name exactly matching "test_cal".

#### 5) Sort by name (A-Z)
**Request:**
`/root/snapshots_history?page=1&per_page=100&sort=name&descending=false`

Returns snapshots sorted alphabetically by name (A to Z).

#### 6) Sort by status (errors first)
**Request:**
`/root/snapshots_history?page=1&per_page=100&sort=status&descending=true`

Returns snapshots sorted by status with errors first, then running,
pending, skipped, and finished last.

**Note:** Cannot use both `name` (exact match) and `name_part` (substring
match) in the same request - this will return a 400 Bad Request error.
"""
if search_filters.name is not None and search_filters.name_part is not None:
raise HTTPException(
status_code=400,
detail=(
"Cannot use both 'name' (exact match) and 'name_part' "
"(substring match) parameters together. Use only one."
),
)
total, snapshots = root.get_latest_snapshots(
pages_filter=page_filter,
search_filter=SearchWithIdFilter(**search_filters.model_dump()),
descending=descending,
)
snapshots_dumped = [
SimplifiedSnapshotWithMetadata(**snapshot.dump().model_dump())
for snapshot in snapshots
]

# Apply sorting if sort field is specified
if sort is not None:

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i think there is a bug since it happens after pagination, it will sort the paginated file not everything

snapshots_dumped = sorted(
snapshots_dumped,
key=lambda s: _get_sort_key(s, sort, descending),
reverse=descending,
)

return PagedCollection[SimplifiedSnapshotWithMetadata](
page=page_filter.page,
per_page=page_filter.per_page,
Expand Down
5 changes: 5 additions & 0 deletions backend/qualibrate_app/api/routes/utils/dependencies.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,10 @@ def get_page_filter(


def get_search_filter(
name: Annotated[
str | None,
Query(description="Exact snapshot name to match."),
] = None,
name_part: Annotated[
str | None,
Query(description="Substring to match within snapshot name."),
Expand All @@ -57,6 +61,7 @@ def get_search_filter(
] = None,
) -> SearchFilter:
return SearchFilter(
name=name,
min_node_id=min_node_id,
max_node_id=max_node_id,
min_date=min_date,
Expand Down
2 changes: 1 addition & 1 deletion test_search_features.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ def get_example_snapshot_id():
try:
items = resp.json().get("items", [])
return items[0]["id"] if items else None
except:
except Exception:
return None


Expand Down
Loading