Coverage for functions \ flipdare \ request \ data \ search_request_adapter.py: 59%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13from __future__ import annotations 

14 

15import hashlib 

16from typing import Any 

17 

18from flipdare.app_log import LOG 

19from flipdare.constants import IS_DEBUG 

20from flipdare.generated.schema.search.search_request_schema import SearchRequestSchema 

21from flipdare.generated.shared.search.search_collections import SearchCollections 

22from flipdare.generated.shared.search.search_obj_type import SearchObjType 

23from flipdare.generated.shared.search.search_relation_type import SearchRelationType 

24from flipdare.generated.shared.search.search_sort_type import SearchSortType 

25from flipdare.request.app_request import AppRequest 

26from flipdare.request.request_adapter import RequestAdapter 

27from flipdare.search.core.query_builder import QueryBuilderFactory, QueryBuilderType 

28 

29__all__ = ["SearchRequestAdapter"] 

30 

31 

32class SearchRequestAdapter(RequestAdapter[SearchRequestSchema]): 

33 __slots__ = () 

34 

35 SCHEMA_CLS = SearchRequestSchema 

36 VALIDATORS = () 

37 

38 def __init__( 

39 self, 

40 request: AppRequest[Any], 

41 ) -> None: 

42 super().__init__(request) 

43 

44 @property 

45 def uid(self) -> str | None: 

46 return self.authenticated_uid 

47 

48 @property 

49 def query_str(self) -> str: 

50 return self.data["query"] 

51 

52 @property 

53 def page(self) -> int: 

54 return self.data["page_num"] 

55 

56 @property 

57 def obj_types(self) -> list[SearchObjType] | None: 

58 values = self.data.get("obj_types", None) 

59 if values is None: 

60 return None 

61 return [SearchObjType(value) for value in values] 

62 

63 @property 

64 def collection(self) -> SearchCollections: 

65 return self.data["collection"] 

66 

67 @property 

68 def relation_type(self) -> SearchRelationType | None: 

69 value = self.data.get("relation_type", None) 

70 if value is None: 

71 return None 

72 return SearchRelationType(value) 

73 

74 @property 

75 def sort_type(self) -> SearchSortType: 

76 return self.data["sort_type"] 

77 

78 @property 

79 def auto_complete(self) -> bool: 

80 return self.data.get("auto_complete", False) 

81 

82 @property 

83 def query(self) -> QueryBuilderType: 

84 query_builder = QueryBuilderFactory.create(self.data, uid=self.uid) 

85 return query_builder.build() 

86 

87 @property 

88 def cache_key(self) -> str | None: 

89 """Generates a unique hash key for the search request based on its parameters.""" 

90 relation_type = self.relation_type 

91 query = self.query_str.strip().lower() 

92 if relation_type is not None: 

93 if IS_DEBUG: 

94 LOG().debug(f"Cant cache specific user requests {relation_type}: Query={query}") 

95 # Don't cache known user specific searches, since they 

96 # are specific to the user (and cant be shared across users) 

97 return None 

98 

99 sort_type = self.sort_type 

100 if not sort_type.can_cache: 

101 if IS_DEBUG: 

102 LOG().debug(f"Cant cache requests with sort type {sort_type}: Query={query}") 

103 return None 

104 

105 obj_types = self.obj_types 

106 obj_types_str = ( 

107 ",".join(sorted([obj_type.value for obj_type in obj_types])) if obj_types else "all" 

108 ) 

109 

110 auto_complete = self.auto_complete 

111 page_num = self.page 

112 

113 # Create the "Intent" key (Ignore page_num) 

114 intent_parts = [ 

115 query, 

116 obj_types_str, 

117 # str(relation_type), # all is assumed. 

118 str(sort_type), 

119 auto_complete, 

120 ] 

121 

122 raw = ":".join(map(str, intent_parts)).encode() 

123 intent_hash = hashlib.md5(raw).hexdigest() # noqa: S324 

124 cache_key = f"{intent_hash}:pg:{page_num}" 

125 if IS_DEBUG: 

126 LOG().debug( 

127 f"Generated cache key for search request: {cache_key} (Intent: {intent_parts})", 

128 ) 

129 

130 return cache_key