diff --git a/src/crud.py b/src/crud.py index 74be881f..73d4b8dc 100644 --- a/src/crud.py +++ b/src/crud.py @@ -1,15 +1,15 @@ import os from collections.abc import Sequence from logging import getLogger -from typing import Optional +from typing import Any, Optional from dotenv import load_dotenv from nanoid import generate as generate_nanoid from openai import AsyncOpenAI -from sqlalchemy import Select, cast, func, insert, select, update +from sqlalchemy import Select, and_, cast, func, insert, not_, or_, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy.types import BigInteger +from sqlalchemy.types import BigInteger, Numeric from src.config import settings @@ -33,6 +33,330 @@ SESSION_PEERS_LIMIT = int(os.getenv("SESSION_PEERS_LIMIT", 10)) # Using OpenAI provider for embeddings as it's the most common embedding_client = ModelClient(provider=ModelProvider.OPENAI) + +def apply_filter(stmt: Select, filter: dict[str, Any] | None = None) -> Select: + """ + Apply advanced filter to a SQL statement based on filter dictionary. + + Supports logical operators (AND, OR, NOT), comparison operators + (gte, lte, gt, lt, ne, contains, icontains, in), and wildcard character (*). + + Examples: + # Simple filters (backward compatible) + {"peer_name": "alice", "metadata": {"type": "user"}} + + # Logical operators + {"AND": [{"peer_name": "alice"}, {"created_at": {"gte": "2024-01-01"}}]} + {"OR": [{"peer_name": "alice"}, {"peer_name": "bob"}]} + {"NOT": [{"peer_name": "alice"}]} + + # Comparison operators + {"created_at": {"gte": "2024-01-01", "lte": "2024-12-31"}} + {"peer_name": {"in": ["alice", "bob"]}} + {"content": {"contains": "hello"}} + + # Wildcards (matches everything for that field) + {"peer_name": "*"} + + Args: + stmt: SQLAlchemy Select statement to modify + filter: Optional filter dictionary + + Returns: + Modified Select statement with filter applied if provided + """ + if filter is None: + return stmt + + # Get the model class from the statement's columns + model_class = stmt.column_descriptions[0]["entity"] + + conditions = _build_filter_conditions(filter, model_class) + if conditions is not None: + stmt = stmt.where(conditions) + + return stmt + + +def _build_filter_conditions(filter_dict: dict[str, Any], model_class) -> Any: + """ + Recursively build filter conditions from a filter dictionary. + + Args: + filter_dict: Filter dictionary that may contain logical operators + model_class: SQLAlchemy model class for column access + + Returns: + SQLAlchemy condition object or None + """ + conditions = [] + + # Handle logical operators + if "AND" in filter_dict: + and_conditions = [] + for sub_filter in filter_dict["AND"]: + sub_condition = _build_filter_conditions(sub_filter, model_class) + if sub_condition is not None: + and_conditions.append(sub_condition) + if and_conditions: + conditions.append(and_(*and_conditions)) + + if "OR" in filter_dict: + or_conditions = [] + for sub_filter in filter_dict["OR"]: + sub_condition = _build_filter_conditions(sub_filter, model_class) + if sub_condition is not None: + or_conditions.append(sub_condition) + if or_conditions: + conditions.append(or_(*or_conditions)) + + if "NOT" in filter_dict: + not_conditions = [] + for sub_filter in filter_dict["NOT"]: + sub_condition = _build_filter_conditions(sub_filter, model_class) + if sub_condition is not None: + not_conditions.append(sub_condition) + if not_conditions: + conditions.append(not_(and_(*not_conditions))) + + # Handle field-level conditions (skip logical operator keys) + logical_keys = {"AND", "OR", "NOT"} + for key, value in filter_dict.items(): + if key in logical_keys: + continue + + condition = _build_field_condition(key, value, model_class) + if condition is not None: + conditions.append(condition) + + # Combine all conditions with AND + if len(conditions) == 0: + return None + elif len(conditions) == 1: + return conditions[0] + else: + return and_(*conditions) + + +def _build_field_condition(key: str, value: Any, model_class) -> Any: + """ + Build a condition for a single field. + + Args: + key: Field name + value: Field value or comparison dict + model_class: SQLAlchemy model class + + Returns: + SQLAlchemy condition object or None + """ + # Map 'metadata' to 'h_metadata' for all models + column_name = "h_metadata" if key == "metadata" else key + + # Check if the column exists on the model + if not hasattr(model_class, column_name): + return None + + column = getattr(model_class, column_name) + + # Handle wildcard - matches everything, so no condition needed + if value == "*": + return None + + # Handle comparison operators vs regular values + if isinstance(value, dict): + # Check if this is a comparison operators dict by looking for known operators + comparison_operators = { + "gte", + "lte", + "gt", + "lt", + "ne", + "in", + "contains", + "icontains", + } + is_comparison_dict = any(key in comparison_operators for key in value) + + if is_comparison_dict: + return _build_comparison_conditions(column, column_name, value) + else: + # This is a regular value that happens to be a dict + # For JSONB metadata, check if it contains nested comparison operators + if column_name == "h_metadata": + return _build_nested_metadata_conditions(column, value, model_class) + else: + return column == value + else: + # Simple equality or contains for JSONB + if column_name == "h_metadata": + return column.contains(value) + else: + return column == value + + +def _build_nested_metadata_conditions( + column, metadata_dict: dict[str, Any], model_class +) -> Any: + """ + Build conditions for nested metadata fields with comparison operators. + + Args: + column: SQLAlchemy JSONB column object + metadata_dict: Dictionary containing nested field conditions + model_class: SQLAlchemy model class + + Returns: + Combined SQLAlchemy condition object or None + """ + conditions = [] + comparison_operators = { + "gte", + "lte", + "gt", + "lt", + "ne", + "in", + "contains", + "icontains", + } + + for field_name, field_value in metadata_dict.items(): + if isinstance(field_value, dict) and any( + op in comparison_operators for op in field_value + ): + # This field has comparison operators + field_conditions = [] + for operator, op_value in field_value.items(): + if op_value == "*": + continue + + condition = None + + if operator == "gte": + # For numeric comparisons, cast to numeric; otherwise use string comparison + if isinstance(op_value, int | float): + field_accessor = cast(column[field_name].astext, Numeric) + condition = field_accessor >= op_value + else: + condition = column[field_name].astext >= str(op_value) + elif operator == "lte": + if isinstance(op_value, int | float): + field_accessor = cast(column[field_name].astext, Numeric) + condition = field_accessor <= op_value + else: + condition = column[field_name].astext <= str(op_value) + elif operator == "gt": + if isinstance(op_value, int | float): + field_accessor = cast(column[field_name].astext, Numeric) + condition = field_accessor > op_value + else: + condition = column[field_name].astext > str(op_value) + elif operator == "lt": + if isinstance(op_value, int | float): + field_accessor = cast(column[field_name].astext, Numeric) + condition = field_accessor < op_value + else: + condition = column[field_name].astext < str(op_value) + elif operator == "ne": + # For ne, we need to handle both numeric and text comparisons + if isinstance(op_value, int | float): + field_accessor = cast(column[field_name].astext, Numeric) + condition = field_accessor != op_value + else: + condition = column[field_name].astext != str(op_value) + elif operator == "in": + if isinstance(op_value, list): + condition = column[field_name].astext.in_( + [str(v) for v in op_value] + ) + elif operator == "contains" or operator == "icontains": + condition = column[field_name].astext.ilike(f"%{op_value}%") + + if condition is not None: + field_conditions.append(condition) + + if field_conditions: + if len(field_conditions) == 1: + conditions.append(field_conditions[0]) + else: + conditions.append(and_(*field_conditions)) + else: + # Regular field equality - use JSONB contains for nested object matching + conditions.append(column.contains({field_name: field_value})) + + # Combine all field conditions with AND + if len(conditions) == 0: + return None + elif len(conditions) == 1: + return conditions[0] + else: + return and_(*conditions) + + +def _build_comparison_conditions( + column, column_name: str, comparisons: dict[str, Any] +) -> Any: + """ + Build comparison conditions for a single column. + + Args: + column: SQLAlchemy column object + column_name: Name of the column + comparisons: Dictionary of comparison operators and values + + Returns: + Combined SQLAlchemy condition object or None + """ + conditions = [] + + for operator, op_value in comparisons.items(): + if op_value == "*": + # Wildcard for this operator - skip condition + continue + + condition = None + + if operator == "gte": + condition = column >= op_value + elif operator == "lte": + condition = column <= op_value + elif operator == "gt": + condition = column > op_value + elif operator == "lt": + condition = column < op_value + elif operator == "ne": + condition = column != op_value + elif operator == "in": + if isinstance(op_value, list): + # Check if the list contains wildcard - if so, skip condition (match all) + if "*" in op_value: + continue + else: + condition = column.in_(op_value) + elif operator == "contains": + if column_name == "h_metadata": + # For JSONB columns, use JSONB contains + condition = column.contains(op_value) + else: + # For text columns, use ILIKE + condition = column.ilike(f"%{op_value}%") + elif operator == "icontains": + # Case-insensitive contains for text columns + condition = column.ilike(f"%{op_value}%") + + if condition is not None: + conditions.append(condition) + + # Combine all conditions for this field with AND + if len(conditions) == 0: + return None + elif len(conditions) == 1: + return conditions[0] + else: + return and_(*conditions) + + ######################################################## # workspace methods ######################################################## @@ -87,8 +411,7 @@ async def get_all_workspaces( filter: Filter the workspaces by a dictionary of metadata """ stmt = select(models.Workspace) - if filter is not None: - stmt = stmt.where(models.Workspace.h_metadata.contains(filter)) + stmt = apply_filter(stmt, filter) stmt = stmt.order_by(models.Workspace.created_at) return stmt @@ -235,8 +558,7 @@ async def get_peers( ) -> Select: stmt = select(models.Peer).where(models.Peer.workspace_name == workspace_name) - if filter is not None: - stmt = stmt.where(models.Peer.h_metadata.contains(filter)) + stmt = apply_filter(stmt, filter) stmt = stmt.order_by(models.Peer.created_at) @@ -312,8 +634,7 @@ async def get_sessions_for_peer( if is_active is not None: stmt = stmt.where(models.Session.is_active == is_active) - if filter is not None: - stmt = stmt.where(models.Session.h_metadata.contains(filter)) + stmt = apply_filter(stmt, filter) stmt = stmt.order_by(models.Session.created_at) @@ -338,8 +659,7 @@ async def get_sessions( if is_active: stmt = stmt.where(models.Session.is_active.is_(True)) - if filter is not None: - stmt = stmt.where(models.Session.h_metadata.contains(filter)) + stmt = apply_filter(stmt, filter) stmt = stmt.order_by(models.Session.created_at) @@ -1275,13 +1595,10 @@ async def get_messages( models.Message.session_name == session_name, ] - # Add metadata filter if provided - if filter is not None: - base_conditions.append(models.Message.h_metadata.contains(filter)) - # Apply message count limit first (takes precedence over token limit) if message_count_limit is not None: stmt = select(models.Message).where(*base_conditions) + stmt = apply_filter(stmt, filter) # For message count limit, we want the most recent N messages # So we order by id desc to get most recent, then apply limit stmt = stmt.order_by(models.Message.id.desc()).limit(message_count_limit) @@ -1291,7 +1608,6 @@ async def get_messages( stmt = stmt.order_by(models.Message.id.desc()) else: stmt = stmt.order_by(models.Message.id.asc()) - elif token_limit is not None: # Apply token limit logic # Create a subquery that calculates running sum of tokens for most recent messages @@ -1312,16 +1628,17 @@ async def get_messages( .join(token_subquery, models.Message.id == token_subquery.c.id) .where(token_subquery.c.running_token_sum <= token_limit) ) + stmt = apply_filter(stmt, filter) # Apply final ordering based on reverse parameter if reverse: stmt = stmt.order_by(models.Message.id.desc()) else: stmt = stmt.order_by(models.Message.id.asc()) - else: # Default case - no limits applied stmt = select(models.Message).where(*base_conditions) + stmt = apply_filter(stmt, filter) if reverse: stmt = stmt.order_by(models.Message.id.desc()) else: @@ -1394,8 +1711,7 @@ async def get_messages_for_peer( .where(models.Message.session_name.is_(None)) ) - if filter is not None: - stmt = stmt.where(models.Message.h_metadata.contains(filter)) + stmt = apply_filter(stmt, filter) if reverse: stmt = stmt.order_by(models.Message.id.desc()) @@ -1535,8 +1851,7 @@ async def query_documents( stmt = stmt.where( models.Document.embedding.cosine_distance(embedding_query) < max_distance ) - if filter is not None: - stmt = stmt.where(models.Document.h_metadata.contains(filter)) + stmt = apply_filter(stmt, filter) stmt = stmt.limit(top_k).order_by( models.Document.embedding.cosine_distance(embedding_query) ) diff --git a/src/routers/messages.py b/src/routers/messages.py index f8dd8df1..04f83925 100644 --- a/src/routers/messages.py +++ b/src/routers/messages.py @@ -4,7 +4,7 @@ from typing import Any, Optional from fastapi import APIRouter, BackgroundTasks, Body, Depends, Path, Query from fastapi_pagination import Page -from fastapi_pagination.ext.sqlalchemy import paginate +from fastapi_pagination.ext.sqlalchemy import apaginate from sqlalchemy.sql import insert from src import crud, schemas @@ -333,7 +333,7 @@ async def get_messages( reverse=reverse, ) - return await paginate(db, messages_query) + return await apaginate(db, messages_query) except ValueError as e: logger.warning(f"Failed to get messages for session {session_id}: {str(e)}") raise ResourceNotFoundException("Session not found") from e diff --git a/src/routers/peers.py b/src/routers/peers.py index 1bd67427..f045fe93 100644 --- a/src/routers/peers.py +++ b/src/routers/peers.py @@ -13,7 +13,7 @@ from fastapi import ( from fastapi.exceptions import HTTPException from fastapi.responses import StreamingResponse from fastapi_pagination import Page -from fastapi_pagination.ext.sqlalchemy import paginate +from fastapi_pagination.ext.sqlalchemy import apaginate from src import agent, crud, schemas from src.dependencies import db @@ -51,7 +51,7 @@ async def get_peers( if filter_param == {}: filter_param = None - return await paginate( + return await apaginate( db, await crud.get_peers(workspace_name=workspace_id, filter=filter_param), ) @@ -138,7 +138,7 @@ async def get_sessions_for_peer( if hasattr(options, "is_active"): is_active = options.is_active - return await paginate( + return await apaginate( db, await crud.get_sessions_for_peer( workspace_name=workspace_id, @@ -286,7 +286,7 @@ async def get_messages_for_peer( reverse=reverse, ) - return await paginate(db, messages_query) + return await apaginate(db, messages_query) except ValueError as e: logger.warning(f"Failed to get messages for peer {peer_id}: {str(e)}") raise ResourceNotFoundException("Peer not found") from e @@ -336,4 +336,4 @@ async def search_peer( """Search a Peer""" stmt = await crud.search(query, workspace_name=workspace_id, peer_name=peer_id) - return await paginate(db, stmt) + return await apaginate(db, stmt) diff --git a/src/routers/sessions.py b/src/routers/sessions.py index 6fb47a8b..2d91598a 100644 --- a/src/routers/sessions.py +++ b/src/routers/sessions.py @@ -3,7 +3,7 @@ from typing import Optional from fastapi import APIRouter, Body, Depends, Path, Query, Response from fastapi_pagination import Page -from fastapi_pagination.ext.sqlalchemy import paginate +from fastapi_pagination.ext.sqlalchemy import apaginate from src import crud, schemas from src.dependencies import db @@ -94,7 +94,7 @@ async def get_sessions( if hasattr(options, "is_active"): # Check if is_active is present is_active_param = options.is_active - return await paginate( + return await apaginate( db, await crud.get_sessions( workspace_name=workspace_id, @@ -365,7 +365,7 @@ async def get_session_peers( peers_query = await crud.get_peers_from_session( workspace_name=workspace_id, session_name=session_name ) - return await paginate(db, peers_query) + return await apaginate(db, peers_query) except ValueError as e: logger.warning(f"Failed to get peers from session {session_name}: {str(e)}") raise ResourceNotFoundException("Session not found") from e @@ -472,4 +472,4 @@ async def search_session( query, workspace_name=workspace_id, session_name=session_id ) - return await paginate(db, stmt) + return await apaginate(db, stmt) diff --git a/src/routers/workspaces.py b/src/routers/workspaces.py index a1aa4524..a6bcbd85 100644 --- a/src/routers/workspaces.py +++ b/src/routers/workspaces.py @@ -3,7 +3,7 @@ from typing import Optional from fastapi import APIRouter, Body, Depends, Path from fastapi_pagination import Page -from fastapi_pagination.ext.sqlalchemy import paginate +from fastapi_pagination.ext.sqlalchemy import apaginate from src import crud, schemas from src.dependencies import db @@ -65,7 +65,7 @@ async def get_all_workspaces( if filter_param == {}: filter_param = None - return await paginate( + return await apaginate( db, await crud.get_all_workspaces(filter=filter_param), ) @@ -104,4 +104,4 @@ async def search_workspace( """Search a Workspace""" stmt = await crud.search(query, workspace_name=workspace_id) - return await paginate(db, stmt) + return await apaginate(db, stmt) diff --git a/tests/routes/test_messages.py b/tests/routes/test_messages.py index 6273ed4c..9458c318 100644 --- a/tests/routes/test_messages.py +++ b/tests/routes/test_messages.py @@ -354,12 +354,12 @@ async def test_get_filtered_messages(client, db_session, sample_data): response = client.post( f"/v2/workspaces/{test_workspace.name}/sessions/{test_session.name}/messages/list", - json={"filter": {"key": "value2"}}, + json={"filter": {"metadata": {"key": "value2"}}}, ) assert response.status_code == 200 data = response.json() assert "items" in data - assert len(data["items"]) > 0 + assert len(data["items"]) == 1 assert data["items"][0]["content"] == "Test message 2" assert data["items"][0]["peer_id"] == test_peer.name assert data["items"][0]["session_id"] == test_session.name @@ -406,10 +406,10 @@ async def test_get_filtered_messages_with_complex_filter( db_session.add(test_message3) await db_session.commit() - # Filter by multiple criteria + # Test old-style filter (backward compatibility) response = client.post( f"/v2/workspaces/{test_workspace.name}/sessions/{test_session.name}/messages/list", - json={"filter": {"priority": "high", "category": "technical"}}, + json={"filter": {"metadata": {"priority": "high", "category": "technical"}}}, ) assert response.status_code == 200 data = response.json() @@ -417,6 +417,40 @@ async def test_get_filtered_messages_with_complex_filter( # Should return messages 1 and 2 (both have high priority and technical category) assert len(data["items"]) >= 2 + # Test new-style filter with AND operator + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{test_session.name}/messages/list", + json={ + "filter": { + "AND": [ + {"metadata": {"priority": "high"}}, + {"metadata": {"category": "technical"}}, + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + assert "items" in data + assert len(data["items"]) == 2 + + # Test OR filter to get high priority OR question type + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{test_session.name}/messages/list", + json={ + "filter": { + "OR": [ + {"metadata": {"priority": "high"}}, + {"metadata": {"type": "question"}}, + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + assert "items" in data + assert len(data["items"]) == 3 # All messages should match + @pytest.mark.asyncio async def test_update_message(client, db_session, sample_data): diff --git a/tests/routes/test_peers.py b/tests/routes/test_peers.py index 84ee9243..bce0e426 100644 --- a/tests/routes/test_peers.py +++ b/tests/routes/test_peers.py @@ -101,17 +101,29 @@ def test_get_peers(client, sample_data): assert "items" in data assert len(data["items"]) > 0 - # Get peers with filter + # Get peers with simple filter (backward compatibility) response = client.post( f"/v2/workspaces/{test_workspace.name}/peers/list", - json={"filter": {"peer_key": "peer_value"}}, + json={"filter": {"metadata": {"peer_key": "peer_value"}}}, ) assert response.status_code == 200 data = response.json() assert "items" in data - assert len(data["items"]) >= 2 + assert len(data["items"]) == 2 assert data["items"][0]["metadata"]["peer_key"] == "peer_value" + # Test new filter with NOT operator + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={"filter": {"NOT": [{"metadata": {"peer_key": "peer_value2"}}]}}, + ) + assert response.status_code == 200 + data = response.json() + assert "items" in data + # Should find peers that don't have peer_key = "peer_value2" + # This includes the 2 peers with "peer_value" + the sample peer with empty metadata + assert len(data["items"]) == 3 + def test_get_peers_with_empty_filter(client, sample_data): """Test peer listing with empty filter object""" diff --git a/tests/routes/test_sessions.py b/tests/routes/test_sessions.py index 75d71973..94b59629 100644 --- a/tests/routes/test_sessions.py +++ b/tests/routes/test_sessions.py @@ -170,12 +170,12 @@ def test_get_sessions(client, sample_data): assert data["workspace_id"] == test_workspace.name response = client.post( f"/v2/workspaces/{test_workspace.name}/sessions/list", - json={"filter": {"test_key": "test_value"}}, + json={"filter": {"metadata": {"test_key": "test_value"}}}, ) assert response.status_code == 200 data = response.json() assert "items" in data - assert len(data["items"]) > 0 + assert len(data["items"]) == 1 assert data["items"][0]["metadata"] == {"test_key": "test_value"} assert data["items"][0]["workspace_id"] == test_workspace.name diff --git a/tests/routes/test_workspaces.py b/tests/routes/test_workspaces.py index 158146c4..1c1d0709 100644 --- a/tests/routes/test_workspaces.py +++ b/tests/routes/test_workspaces.py @@ -88,12 +88,12 @@ async def test_get_all_workspaces(client, db_session, sample_data): response = client.post( "/v2/workspaces/list", - json={"filter": {"test_key": "test_value"}}, + json={"filter": {"metadata": {"test_key": "test_value"}}}, ) assert response.status_code == 200 data = response.json() assert "items" in data - assert len(data["items"]) > 0 + assert len(data["items"]) == 1 assert data["items"][0]["metadata"] == {"test_key": "test_value"} diff --git a/tests/test_advanced_filters.py b/tests/test_advanced_filters.py new file mode 100644 index 00000000..261ca19b --- /dev/null +++ b/tests/test_advanced_filters.py @@ -0,0 +1,593 @@ +""" +Tests for advanced filter functionality including logical operators, +comparison operators, and wildcards across multiple models. +""" + +import pytest +from fastapi.testclient import TestClient +from nanoid import generate as generate_nanoid +from sqlalchemy.ext.asyncio import AsyncSession + +from src.models import Peer, Workspace + + +@pytest.mark.asyncio +async def test_logical_operators_and_filters( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test AND, OR, NOT logical operators in filters""" + test_workspace, test_peer = sample_data + + # Create multiple peers with different metadata + peer1_name = str(generate_nanoid()) + peer2_name = str(generate_nanoid()) + peer3_name = str(generate_nanoid()) + + client.post( + f"/v2/workspaces/{test_workspace.name}/peers", + json={ + "name": peer1_name, + "metadata": { + "role": "admin", + "department": "engineering", + "level": "senior", + }, + }, + ) + client.post( + f"/v2/workspaces/{test_workspace.name}/peers", + json={ + "name": peer2_name, + "metadata": { + "role": "user", + "department": "engineering", + "level": "junior", + }, + }, + ) + client.post( + f"/v2/workspaces/{test_workspace.name}/peers", + json={ + "name": peer3_name, + "metadata": {"role": "admin", "department": "sales", "level": "senior"}, + }, + ) + + # Test AND operator - peers who are admin AND in engineering + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={ + "filter": { + "AND": [ + {"metadata": {"role": "admin"}}, + {"metadata": {"department": "engineering"}}, + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + assert data["items"][0]["id"] == peer1_name + + # Test OR operator - peers who are admin OR in engineering + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={ + "filter": { + "OR": [ + {"metadata": {"role": "admin"}}, + {"metadata": {"department": "engineering"}}, + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 3 # All three peers match + + # Test NOT operator - peers who are NOT admin + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={"filter": {"NOT": [{"metadata": {"role": "admin"}}]}}, + ) + assert response.status_code == 200 + data = response.json() + found_names = [item["id"] for item in data["items"]] + assert peer2_name in found_names # Only peer2 is not admin + assert peer1_name not in found_names + assert peer3_name not in found_names + + +@pytest.mark.asyncio +async def test_comparison_operators_filters( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test comparison operators (gte, lte, gt, lt, ne, in, contains, icontains)""" + test_workspace, test_peer = sample_data + + # Create session with messages containing different metadata + session_id = str(generate_nanoid()) + session_response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions", + json={"id": session_id, "peer_names": {test_peer.name: {}}}, + ) + assert session_response.status_code == 200 + + # Create messages with numeric metadata for comparison tests + messages_response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages", + json={ + "messages": [ + { + "content": "Message with score 10", + "peer_id": test_peer.name, + "metadata": { + "score": 10, + "category": "high", + "tags": ["important", "urgent"], + }, + }, + { + "content": "Message with score 5", + "peer_id": test_peer.name, + "metadata": {"score": 5, "category": "medium", "tags": ["normal"]}, + }, + { + "content": "Message with score 1", + "peer_id": test_peer.name, + "metadata": {"score": 1, "category": "low", "tags": ["minor"]}, + }, + ] + }, + ) + assert messages_response.status_code == 200 + + # Test gte (greater than or equal) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list", + json={"filter": {"metadata": {"score": {"gte": 5}}}}, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 2 # Messages with score 10 and 5 + + # Test lte (less than or equal) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list", + json={"filter": {"metadata": {"score": {"lte": 5}}}}, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 2 # Messages with score 5 and 1 + + # Test gt (greater than) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list", + json={"filter": {"metadata": {"score": {"gt": 5}}}}, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 # Only message with score 10 + + # Test lt (less than) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list", + json={"filter": {"metadata": {"score": {"lt": 5}}}}, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 # Only message with score 1 + + # Test ne (not equal) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list", + json={"filter": {"metadata": {"score": {"ne": 5}}}}, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 2 # Messages with score 10 and 1 + + # Test in operator + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list", + json={"filter": {"metadata": {"category": {"in": ["high", "low"]}}}}, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 2 # Messages with category "high" and "low" + + # Test contains operator for text content + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list", + json={"filter": {"content": {"contains": "score 10"}}}, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + assert "score 10" in data["items"][0]["content"] + + # Test icontains operator (case-insensitive) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list", + json={"filter": {"content": {"icontains": "MESSAGE"}}}, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 3 # All messages contain "message" (case-insensitive) + + +@pytest.mark.asyncio +async def test_wildcard_filters( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test wildcard (*) filters that match everything for a field""" + test_workspace, test_peer = sample_data + + # Create peers with different names + peer1_name = str(generate_nanoid()) + peer2_name = str(generate_nanoid()) + + client.post( + f"/v2/workspaces/{test_workspace.name}/peers", + json={"name": peer1_name, "metadata": {"type": "bot"}}, + ) + client.post( + f"/v2/workspaces/{test_workspace.name}/peers", + json={"name": peer2_name, "metadata": {"type": "human"}}, + ) + + # Test wildcard for name field - should match all peers regardless of name + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={ + "filter": { + "AND": [ + {"name": "*"}, # Wildcard matches all names + {"metadata": {"type": "bot"}}, + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + # Should find the bot peer (wildcard doesn't filter anything) + found_names = [item["id"] for item in data["items"]] + assert peer1_name in found_names + + # Test wildcard in comparison operators + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={ + "filter": { + "name": {"in": ["*"]} # Wildcard in comparison should also match all + } + }, + ) + assert response.status_code == 200 + data = response.json() + # Since wildcard should be ignored, this should return all peers + assert len(data["items"]) >= 3 # At least the 3 peers we know about + + +@pytest.mark.asyncio +async def test_complex_nested_filters( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test complex nested logical operations""" + test_workspace, test_peer = sample_data + + # Create session and messages for complex filtering + session_id = str(generate_nanoid()) + client.post( + f"/v2/workspaces/{test_workspace.name}/sessions", + json={"id": session_id, "peer_names": {test_peer.name: {}}}, + ) + + # Create messages with various metadata combinations + client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages", + json={ + "messages": [ + { + "content": "Priority urgent task", + "peer_id": test_peer.name, + "metadata": { + "priority": "urgent", + "status": "open", + "assignee": "alice", + }, + }, + { + "content": "Normal task for bob", + "peer_id": test_peer.name, + "metadata": { + "priority": "normal", + "status": "open", + "assignee": "bob", + }, + }, + { + "content": "Completed urgent task", + "peer_id": test_peer.name, + "metadata": { + "priority": "urgent", + "status": "completed", + "assignee": "alice", + }, + }, + { + "content": "Low priority task", + "peer_id": test_peer.name, + "metadata": { + "priority": "low", + "status": "open", + "assignee": "charlie", + }, + }, + ] + }, + ) + + # Complex filter: (urgent OR normal priority) AND open status AND NOT assigned to charlie + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list", + json={ + "filter": { + "AND": [ + { + "OR": [ + {"metadata": {"priority": "urgent"}}, + {"metadata": {"priority": "normal"}}, + ] + }, + {"metadata": {"status": "open"}}, + {"NOT": [{"metadata": {"assignee": "charlie"}}]}, + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 2 # Should match first two messages + + # Verify the correct messages were returned + contents = [item["content"] for item in data["items"]] + assert "Priority urgent task" in contents + assert "Normal task for bob" in contents + assert "Completed urgent task" not in contents # Wrong status + assert "Low priority task" not in contents # Wrong assignee and priority + + +@pytest.mark.asyncio +async def test_filters_across_different_models( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test that filters work consistently across different models""" + test_workspace, test_peer = sample_data + + # Test workspace filters + workspace_name = str(generate_nanoid()) + client.post( + "/v2/workspaces", + json={ + "name": workspace_name, + "metadata": {"environment": "production", "version": "2.0"}, + }, + ) + + response = client.post( + "/v2/workspaces/list", + json={ + "filter": { + "AND": [ + {"metadata": {"environment": "production"}}, + {"metadata": {"version": {"gte": "2.0"}}}, + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + found_names = [item["id"] for item in data["items"]] + assert workspace_name in found_names + + # Test session filters with comparison operators + session1_id = str(generate_nanoid()) + session2_id = str(generate_nanoid()) + + client.post( + f"/v2/workspaces/{test_workspace.name}/sessions", + json={ + "id": session1_id, + "peer_names": {test_peer.name: {}}, + "metadata": {"duration": 30, "type": "meeting"}, + }, + ) + client.post( + f"/v2/workspaces/{test_workspace.name}/sessions", + json={ + "id": session2_id, + "peer_names": {test_peer.name: {}}, + "metadata": {"duration": 60, "type": "workshop"}, + }, + ) + + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/list", + json={ + "filter": { + "OR": [ + {"metadata": {"duration": {"gte": 45}}}, + {"metadata": {"type": "meeting"}}, + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + found_sessions = [item["id"] for item in data["items"]] + assert session1_id in found_sessions # Matches type=meeting + assert session2_id in found_sessions # Matches duration>=45 + + +@pytest.mark.asyncio +async def test_filter_edge_cases( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test edge cases and error handling for filters""" + test_workspace, test_peer = sample_data + + # Test empty logical operators + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={ + "filter": { + "AND": [] # Empty AND should not crash + } + }, + ) + assert response.status_code == 200 + + # Test nested empty operators + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={"filter": {"OR": [{"AND": []}, {"name": test_peer.name}]}}, + ) + assert response.status_code == 200 + + # Test filter with non-existent columns (should be ignored gracefully) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={"filter": {"non_existent_column": "value"}}, + ) + assert response.status_code == 200 # Should not crash + + # Test mixed wildcards and regular values + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={ + "filter": { + "AND": [ + {"name": "*"}, # Wildcard + {"name": test_peer.name}, # Regular value + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + # Should find the specific peer since AND combines conditions + found_names = [item["id"] for item in data["items"]] + assert test_peer.name in found_names + + +@pytest.mark.asyncio +async def test_backward_compatibility( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test that old simple filter format still works""" + test_workspace, test_peer = sample_data + + # Create peer with metadata + peer_name = str(generate_nanoid()) + client.post( + f"/v2/workspaces/{test_workspace.name}/peers", + json={ + "name": peer_name, + "metadata": {"role": "admin", "department": "engineering"}, + }, + ) + + # Test old-style simple equality filter + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={"filter": {"metadata": {"role": "admin"}}}, + ) + assert response.status_code == 200 + data = response.json() + found_names = [item["id"] for item in data["items"]] + assert peer_name in found_names + + # Test multiple field simple filter (implicit AND) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={"filter": {"metadata": {"role": "admin"}, "name": peer_name}}, + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + assert data["items"][0]["id"] == peer_name + + +@pytest.mark.asyncio +async def test_range_queries_with_dates( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test range queries that might be used with date fields""" + test_workspace, test_peer = sample_data + + # Create sessions with date-like metadata + session1_id = str(generate_nanoid()) + session2_id = str(generate_nanoid()) + session3_id = str(generate_nanoid()) + + client.post( + f"/v2/workspaces/{test_workspace.name}/sessions", + json={ + "id": session1_id, + "peer_names": {test_peer.name: {}}, + "metadata": {"created_date": "2024-01-15", "priority": 5}, + }, + ) + client.post( + f"/v2/workspaces/{test_workspace.name}/sessions", + json={ + "id": session2_id, + "peer_names": {test_peer.name: {}}, + "metadata": {"created_date": "2024-02-20", "priority": 3}, + }, + ) + client.post( + f"/v2/workspaces/{test_workspace.name}/sessions", + json={ + "id": session3_id, + "peer_names": {test_peer.name: {}}, + "metadata": {"created_date": "2024-03-10", "priority": 8}, + }, + ) + + # Test date range query + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/list", + json={ + "filter": { + "metadata": {"created_date": {"gte": "2024-02-01", "lte": "2024-02-28"}} + } + }, + ) + assert response.status_code == 200 + data = response.json() + found_sessions = [item["id"] for item in data["items"]] + assert session2_id in found_sessions + assert session1_id not in found_sessions + assert session3_id not in found_sessions + + # Test combining date and numeric filters + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/list", + json={ + "filter": { + "AND": [ + {"metadata": {"created_date": {"gte": "2024-01-01"}}}, + {"metadata": {"priority": {"gt": 4}}}, + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + found_sessions = [item["id"] for item in data["items"]] + assert session1_id in found_sessions # priority 5 + assert session3_id in found_sessions # priority 8 + assert session2_id not in found_sessions # priority 3