Coverage for postrfp/ref/service/search_service.py: 89%

1""" 

2DB helper functions for working with search 

3""" 

4 

5from typing import Optional, Any 

6from sqlalchemy.orm import Session 

7from sqlalchemy import or_ 

8from sqlalchemy.sql import func 

9 

10from postrfp.shared.pager import Pager 

11from postrfp.model import ( 

12 Content, 

13 content_subjects_table, 

14) 

15 

16 

17def search_content_by_subject( 

18 session: Session, 

19 user_org_id: str, 

20 subject_id: Optional[int] = None, 

21 search_term: Optional[str] = None, 

22 pager: Optional[Pager] = None, 

23) -> tuple[int, list[Any]]: 

24 """ 

25 Search for content by subject with permission filtering 

26 

27 Returns content items matching the search criteria that the user 

28 has permission to view based on their organization. 

29 

30 Args: 

31 session: Database session 

32 user_org_id: Organization ID of the requesting user 

33 subject_id: Optional subject ID to filter content by 

34 name: Optional search term to match against title or content 

35 pager: Pagination settings 

36 

37 Returns: 

38 Tuple of (total_count, list_of_content_summary_rows) 

39 """ 

40 # Build a query that selects only the columns we need 

41 query = session.query( 

42 Content.id, 

43 Content.schema_id, 

44 Content.title, 

45 Content.date_created, 

46 Content.date_updated, 

47 ).filter( 

48 or_( 

49 # Content authored by user's org 

50 Content.author_org_id == user_org_id 

51 ) 

52 ) 

53 

54 # Filter by subject if provided 

55 if subject_id: 

56 query = query.join( 

57 content_subjects_table, 

58 Content.id == content_subjects_table.c.content_id, 

59 ).filter(content_subjects_table.c.subject_id == subject_id) 

60 

61 # Filter by search term if provided 

62 if search_term: 

63 # Use match search for content + title search for exact matches 

64 query = query.filter( 

65 or_( 

66 # Match against full-text indexed content 

67 func.match(Content.content_fts).against(search_term), 

68 # Also match title for more precise results 

69 Content.title.ilike(f"%{search_term}%"), 

70 ) 

71 ) 

72 

73 # Order by recently updated 

74 query = query.order_by(Content.date_updated.desc()) 

75 

76 # Get total count with same filters but without projection 

77 count_query = query.with_entities(func.count()).order_by(None) 

78 total_records = count_query.scalar() 

79 

80 # Apply pagination 

81 if pager: 

82 query = query.slice(pager.startfrom, pager.goto) 

83 

84 # Execute query to get rows 

85 records = query.all() 

86 

87 return total_records, records