Coverage for postrfp/shared/fetch/searchq.py: 100%

1from typing import ( 

2 List, 

3 Sequence, 

4 Dict, 

5 Any, 

6) 

7 

8from sqlalchemy import text 

9from sqlalchemy.orm import Session 

10 

11from postrfp.model.questionnaire.b36 import from_b36 

12 

13from postrfp.model import misc # nopep8 # noqa: F403 

14from postrfp.templates import get_template # nopep8 # noqa: F403 

15 

16 

17def search( 

18 session: Session, 

19 org_id: str, 

20 search_term: str, 

21 search_options: Sequence[str], 

22 project_id: int, 

23 offset: int, 

24) -> List[Dict[str, Any]]: 

25 """ 

26 Run a search query. 

27 'search_options' must be a sequence of strings. valid values: 

28 "answers", "questions", "notes", "scoreComments" 

29 """ 

30 available_options = {"answers", "questions", "notes", "scoreComments"} 

31 

32 if not set(search_options) <= available_options: 

33 raise ValueError( 

34 f"search_options {search_options} contains values not in {available_options}" 

35 ) 

36 

37 tmpl = get_template("sql/evaluator_search.sql") 

38 sql = tmpl.render(options=set(search_options), project_id=project_id) 

39 sql_text = text(sql) 

40 params = { 

41 "orgId": org_id, 

42 "search_term": misc.clean_search_term(search_term), 

43 "offset": int(offset), 

44 } 

45 hit_list = [] 

46 for row in session.execute(sql_text, params=params).fetchall(): 

47 doc = dict(row._mapping) 

48 try: 

49 doc["object_ref"] = from_b36(row.object_ref) 

50 except TypeError: 

51 pass 

52 hit_list.append(doc) 

53 return hit_list