Merge pull request #885 from SuffolkLITLab/filter-sessions-by-limit

Add session search feature to interview list page; allow limiting session searching by arbitrary metadata column

docassemble/AssemblyLine/data/questions/demo_search_sessions.yml

@@ -25,9 +25,20 @@ subquestion: |
   Searching is not case sensitive.
 fields:
   - Keyword: search_keyword
+  - Limit metadata column: limit_column_name
+    required: False
+  - To value: limit_column_value
+    required: False  
 ---
 code: |
-  matching_results = find_matching_sessions(search_keyword, user_id="all")
+  if limit_column_name:
+    matching_results = find_matching_sessions(
+      search_keyword, user_id="all", metadata_filters = {
+        limit_column_name: (limit_column_value, "ILIKE")
+      }
+      )
+  else:
+    matching_results = find_matching_sessions(search_keyword, user_id="all")
 ---
 event: no_matches
 question: |

docassemble/AssemblyLine/data/questions/interview_list.yml

@@ -132,6 +132,7 @@ code: |
 sections:
   - section_in_progress_forms: In progress forms
   - section_answer_sets: Answer sets
+  - section_search: Search
 progressive: False
 ---
 # A code block is the only way to show navigation with custom labels (mako isn't allowed in `sections`)
@@ -144,7 +145,10 @@ code: |
         },
         {
           "section_answer_sets": ANSWER_SETS_TITLE or word("Answer sets")
-        }
+        },
+        {
+          "section_search": word("Search")
+        },
         ]
   )
 ---
@@ -261,6 +265,10 @@ subquestion: |
   documents.
   % endif
 
+  % if showifdef("limit_filename") or showifdef("search_keyword"):
+  ${ session_list_html(answers = find_matching_sessions(filenames={limit_filename} if limit_filename else None, keyword=search_keyword), filename=None, limit=20, offset=session_page*20, exclude_filenames=al_sessions_to_exclude_from_interview_list, exclude_newly_started_sessions=True) }
+
+  % else:
   ${ session_list_html(filename=None, limit=20, offset=session_page*20, exclude_filenames=al_sessions_to_exclude_from_interview_list, exclude_newly_started_sessions=True) }
 
   <nav aria-label="Page navigation">
@@ -273,8 +281,12 @@ subquestion: |
   % endif
     </ul>
   </nav>  
-  % else:
-  You do not have any current forms in progress
+  % endif
+
+  % endif
+
+  % if showifdef("limit_filename") or showifdef("search_keyword"):
+  ${ action_button_html(url_action("delete_results_action"), label=word("Clear search results"), size="md", color="secondary", icon="times-circle", id_tag="al-clear-search-results") }
   % endif
 # TODO: might be able to save some DB queries here but performance is good
 section: section_in_progress_forms
@@ -304,6 +316,25 @@ script:
       });
   </script>
 ---
+continue button field: section_search
+section: section_search
+question: |
+  Search
+  % if PAGE_QUESTION:
+  ${ str(PAGE_QUESTION).lower() }
+  % else:
+  in progress forms 
+  % endif
+subquestion: |
+  Use a keyword to find results that match the title or description of the session.
+fields:
+  - Limit by form title (optional): limit_filename
+    required: False
+    code: |
+      get_combined_filename_list()
+  - Search term (optional): search_keyword
+    required: False
+---
 event: section_answer_sets
 id: answer set list
 question: |
@@ -441,3 +472,8 @@ code: |
           ),
       )
   )
+---
+event: delete_results_action
+code: |
+  undefine("limit_filename")
+  undefine("search_keyword")

docassemble/AssemblyLine/requirements.txt

@@ -5,4 +5,5 @@ pandas-stubs
 sqlalchemy[mypy]
 types-PyYAML
 PyGithub
-types-requests
+types-requests
+types-psycopg2

docassemble/AssemblyLine/sessions.py

@@ -1,5 +1,8 @@
+import psycopg2
+from psycopg2.extras import DictCursor
+
 from collections.abc import Iterable
-from typing import List, Dict, Any, Optional, Set, Union, Optional
+from typing import List, Dict, Any, Optional, Set, Union, Optional, Tuple
 from docassemble.base.util import (
     all_variables,
     as_datetime,
@@ -73,6 +76,8 @@
     "session_list_html",
     "set_current_session_metadata",
     "set_interview_metadata",
+    "get_filenames_having_sessions",
+    "get_combined_filename_list",
 ]
 
 db = init_sqlalchemy()
@@ -471,9 +476,10 @@ def find_matching_sessions(
     exclude_filenames: Optional[List[str]] = None,
     exclude_newly_started_sessions: bool = False,
     global_search_allowed_roles: Optional[Union[Set[str], List[str]]] = None,
+    metadata_filters: Optional[Dict[str, Tuple[Any, str, Optional[str]]]] = None,
 ) -> List[Dict[str, Any]]:
-    """Get a list of sessions where the metadata for the session matches the provided keyword search terms.
-    This function is designed to be used in a search interface where the user can search for sessions by keyword.
+    """Get a list of sessions where the metadata for the session matches the provided keyword search terms and metadata filters.
+    This function is designed to be used in a search interface where the user can search for sessions by keyword and specific metadata values.
     The keyword search is case-insensitive and will match any part of the metadata column values.
 
     Args:
@@ -489,15 +495,30 @@ def find_matching_sessions(
         exclude_filenames (Optional[List[str]], optional): A list of filenames to exclude from the results. Defaults to None.
         exclude_newly_started_sessions (bool, optional): Whether to exclude sessions that are still on "step 1". Defaults to False.
         global_search_allowed_roles (Union[Set[str],List[str]], optional): A list or set of roles that are allowed to search all sessions. Defaults to {'admin','developer', 'advocate'}. 'admin' and 'developer' are always allowed to search all sessions.
+        metadata_filters (Optional[Dict[str, Tuple[Any, str, Optional[str]]]], optional): A dictionary of metadata column names and their corresponding filter tuples.
+            Each tuple should contain (value, operator, cast_type).
+            - value: The value to compare against
+            - operator: One of '=', '!=', '<', '<=', '>', '>=', 'LIKE', 'ILIKE'
+            - cast_type: Optional. One of 'int', 'float', or None for string (default)
 
     Returns:
-        List[Dict[str, Any]]: A list of saved sessions for the specified filename that match the search keyword
+        List[Dict[str, Any]]: A list of saved sessions for the specified filename that match the search keyword and metadata filters
+
+    Example:
+        matching_sessions = find_matching_sessions(
+            "smith",
+            user_id="all",
+            filenames=[f"{user_info().package}:intake.yml", "docassemble.MyPackage:intake.yml"],
+            metadata_filters={
+                "owner": ("samantha", "ILIKE", None),
+                "age": (30, ">=", "int"),
+                "status": ("%complete%", "LIKE", None)
+            }
+        )
 
     Example:
+        {"owner": ("samantha", "ILIKE", None), "age": (30, ">=", "int"), "status": ("%complete%", "LIKE", None)}
 
-        ```python
-        matching_sessions=find_matching_sessions("smith", user_id="all", filenames=[f"{user_info().package}:intake.yml", "docassemble.MyPackage:intake.yml"])
-        ```
     """
     if not metadata_column_names:
         metadata_column_names = {"title", "auto_title", "description"}
@@ -516,6 +537,32 @@ def find_matching_sessions(
     else:
         metadata_search_conditions = "TRUE"
 
+    # Add metadata filters
+    if metadata_filters:
+        metadata_filter_conditions = []
+        for column, val_tuple in metadata_filters.items():
+            if len(val_tuple) == 2:
+                value, operator = val_tuple
+                cast_type = None
+            else:
+                value, operator, cast_type = val_tuple
+            if cast_type:
+                column_expr = f"CAST(jsonstorage.data->>{repr(column)} AS {cast_type})"
+            else:
+                column_expr = f"jsonstorage.data->>{repr(column)}"
+
+            if operator.upper() in ("LIKE", "ILIKE"):
+                condition = f"COALESCE({column_expr}, '') {operator} :{column}_filter"
+            else:
+                condition = f"{column_expr} {operator} :{column}_filter"
+
+            metadata_filter_conditions.append(condition)
+
+        metadata_filter_sql = " AND ".join(metadata_filter_conditions)
+        metadata_search_conditions = (
+            f"({metadata_search_conditions}) AND ({metadata_filter_sql})"
+        )
+
     # we retrieve the default metadata columns even if we don't search them
     metadata_column_names = set(metadata_column_names).union(
         {"title", "auto_title", "description"}
@@ -605,6 +652,11 @@ def find_matching_sessions(
         for i, filename in enumerate(filenames):
             parameters[f"filename{i}"] = filename
 
+    # Add metadata filter parameters
+    if metadata_filters:
+        for column, val_tuple in metadata_filters.items():
+            parameters[f"{column}_filter"] = val_tuple[0]
+
     with db.connect() as con:
         rs = con.execute(get_sessions_query, parameters)
 
@@ -939,6 +991,7 @@ def session_list_html(
     show_copy_button: bool = True,
     limit: int = 50,
     offset: int = 0,
+    answers: Optional[List[Dict[str, Any]]] = None,
 ) -> str:
     """Return a string containing an HTML-formatted table with the list of user sessions.
     While interview_list_html() is for answer sets, this feature is for standard
@@ -966,25 +1019,24 @@ def session_list_html(
         show_copy_button (bool, optional): If True, show a copy button for answer sets. Defaults to True.
         limit (int, optional): Limit for the number of sessions returned. Defaults to 50.
         offset (int, optional): Offset for the session list. Defaults to 0.
+        answers (Optional[List[Dict[str, Any]], optional): A list of answers to format and display. Defaults to showing all sessions for the current user.
 
 
     Returns:
         str: HTML-formatted table containing the list of user sessions.
     """
-
-    # TODO: think through how to translate this function. Templates probably work best but aren't
-    # convenient to pass around
-    answers = get_saved_interview_list(
-        filename=filename,
-        user_id=user_id,
-        metadata_key_name=metadata_key_name,
-        limit=limit,
-        offset=offset,
-        filename_to_exclude=filename_to_exclude,
-        exclude_current_filename=exclude_current_filename,
-        exclude_filenames=exclude_filenames,
-        exclude_newly_started_sessions=exclude_newly_started_sessions,
-    )
+    if not answers:
+        answers = get_saved_interview_list(
+            filename=filename,
+            user_id=user_id,
+            metadata_key_name=metadata_key_name,
+            limit=limit,
+            offset=offset,
+            filename_to_exclude=filename_to_exclude,
+            exclude_current_filename=exclude_current_filename,
+            exclude_filenames=exclude_filenames,
+            exclude_newly_started_sessions=exclude_newly_started_sessions,
+        )
 
     if not answers:
         return ""
@@ -1536,3 +1588,112 @@ def config_with_language_fallback(
             return interview_list_config.get(config_key)
     else:
         return get_config(top_level_config_key or config_key)
+
+
+def get_filenames_having_sessions(
+    user_id: Optional[Union[int, str]] = None,
+    global_search_allowed_roles: Optional[Union[Set[str], List[str]]] = None,
+) -> List[str]:
+    """Get a list of all filenames that have sessions saved for a given user, in order
+    to help show the user a good list of interviews to filter search results.
+
+    Args:
+        user_id (Optional[Union[int, str]], optional): User ID to get the list of filenames for. Defaults to current logged-in user. Use "all" to get all filenames.
+        global_search_allowed_roles (Optional[Union[Set[str], List[str]]], optional): Roles that are allowed to search for all sessions. Defaults to admin, developer, and advocate.
+
+    Returns:
+        List[str]: List of filenames that have sessions saved for the user.
+    """
+    if not global_search_allowed_roles:
+        global_search_allowed_roles = {"admin", "developer", "advocate"}
+    global_search_allowed_roles = set(global_search_allowed_roles).union(
+        {"admin", "developer"}
+    )
+
+    conn = variables_snapshot_connection()
+    if user_id is None:
+        if user_logged_in():
+            user_id = user_info().id
+        else:
+            log("Asked to get interview list for user that is not logged in")
+            return []
+
+    if user_id == "all":
+        if user_has_privilege(global_search_allowed_roles):
+            user_id = None
+        elif user_logged_in():
+            user_id = user_info().id
+            log(
+                f"User {user_info().email} does not have permission to list interview sessions belonging to other users"
+            )
+        else:
+            log("Asked to get interview list for user that is not logged in")
+            return []
+
+    try:
+        with conn.cursor(cursor_factory=DictCursor) as cur:
+            if user_id is None:
+                query = """
+                    SELECT DISTINCT(filename) AS filename 
+                    FROM userdict
+                """
+                cur.execute(query)
+            else:
+                query = """
+                    SELECT DISTINCT(filename) AS filename 
+                    FROM userdict 
+                    WHERE (%(user_id)s is null OR user_id = %(user_id)s)
+                """
+                cur.execute(query, {"user_id": user_id})
+
+            results = [record["filename"] for record in cur]
+    finally:
+        conn.close()
+
+    return results
+
+
+def get_combined_filename_list(
+    user_id: Optional[Union[int, str]] = None,
+    global_search_allowed_roles: Optional[Union[Set[str], List[str]]] = None,
+) -> List[Dict[str, str]]:
+    """
+    Get a list of all filenames that have sessions saved for a given user. If it is possible
+    to show a descriptive name for the filename (from the main dispatch area of the configuration),
+    it will show that instead of the filename.
+
+    The results will be in the form of [{filename: Descriptive name}], which is what the Docassemble
+    radio button and dropdown list expect.
+
+    Args:
+        user_id (Optional[Union[int, str]], optional): User ID to get the list of filenames for. Defaults to current logged in user. Use "all" to get all filenames.
+        global_search_allowed_roles (Optional[Union[Set[str], List[str]]], optional): Roles that are allowed to search for all sessions. Defaults to admin, developer, and advocate.
+
+    Returns:
+        List[Dict[str, str]]: List of filenames that have sessions saved for the user.
+    """
+    if not global_search_allowed_roles:
+        global_search_allowed_roles = {"admin", "developer", "advocate"}
+    global_search_allowed_roles = set(global_search_allowed_roles).union(
+        {"admin", "developer"}
+    )
+
+    users_filenames = get_filenames_having_sessions(user_id=user_id)
+    interview_filenames = interview_menu()
+    combined_interviews = []
+    for user_interview in users_filenames:
+        found_match = False
+        for interview in interview_filenames:
+            if interview["filename"] == user_interview:
+                combined_interviews.append(
+                    {
+                        interview["filename"]: interview.get(
+                            "title", interview["filename"]
+                        )
+                    }
+                )
+                found_match = True
+                continue
+        if not found_match:
+            combined_interviews.append({user_interview: user_interview})
+    return combined_interviews