Coverage for functions \ flipdare \ search \ core \ filter \ _complex_filter.py: 89%

65 statements  

« prev     ^ index     » next       coverage.py v7.13.0, created at 2026-05-08 12:22 +1000

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 

13 

14from __future__ import annotations 

15 

16from abc import ABC, abstractmethod 

17from dataclasses import dataclass 

18from enum import StrEnum 

19from typing import TYPE_CHECKING, override 

20 

21from flipdare.app_log import LOG 

22from flipdare.constants import IS_TRACE 

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

24from flipdare.search.core.filter._simple_filter import FilterType, SimpleFilter 

25 

26if TYPE_CHECKING: 

27 

28 from flipdare.search.db.app_friend_search import AppFriendSearch 

29 

30__all__ = ["ComplexFilter", "Relationship"] 

31 

32 

33@dataclass 

34class Relationship: 

35 uid: str 

36 relation_type: SearchRelationType 

37 

38 

39class ComplexFilter[T: StrEnum](SimpleFilter[T], ABC): 

40 __slots__ = ("_obj_type", "_prepare_called", "_relationship") 

41 

42 def __init__( 

43 self, 

44 relationship: Relationship | None = None, 

45 filters: dict[T, FilterType] | None = None, 

46 ) -> None: 

47 super().__init__(filters=filters) 

48 self._relationship = relationship 

49 self._prepare_called = False 

50 

51 def _extra_filters(self) -> dict[str, FilterType]: 

52 """Hook for subclasses to inject additional filter key/value pairs.""" 

53 return {} 

54 

55 @abstractmethod 

56 def _prepare(self, friend_search: AppFriendSearch) -> None: 

57 """Hook for filters to perform pre-search logic (e.g., fetching IDs).""" 

58 ... 

59 

60 def prepare(self, friend_search: AppFriendSearch) -> None: 

61 """Hook for filters to perform pre-search logic (e.g., fetching IDs).""" 

62 self._prepare(friend_search) 

63 self._prepare_called = True 

64 

65 @property 

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

67 return self._relationship.uid if self._relationship else None 

68 

69 @property 

70 def relationship(self) -> Relationship | None: 

71 return self._relationship 

72 

73 @property 

74 @override 

75 def filters(self) -> str | None: 

76 if not self._prepare_called: 

77 raise RuntimeError("Filter.prepare() must be called before accessing filter string.") 

78 

79 base_filter = self.filter_by_dict 

80 extra = self._extra_filters() 

81 all_filter = (base_filter or {}) | extra 

82 if IS_TRACE: 

83 LOG().trace( 

84 f"Building Filter with:\n" 

85 f"\tBase Filter: {base_filter}\n" 

86 f"\tExtra: {extra}\n" 

87 f"\tCombined: {all_filter}" 

88 ) 

89 return self._build_filter(all_filter) 

90 

91 @property 

92 @override 

93 def filter_by_dict(self) -> dict[str, FilterType] | None: 

94 filter_by = self._filters 

95 if filter_by is None: 

96 return None 

97 

98 return {key.value: value for key, value in filter_by.items()} 

99 

100 @override 

101 def __str__(self) -> str: 

102 return f"ComplexFilter(uid={self.uid}, relationship={self._relationship}, filters={self._filters})" 

103 

104 @override 

105 def __repr__(self) -> str: 

106 return self.__str__() 

107 

108 @override 

109 def __eq__(self, other: object) -> bool: 

110 if not isinstance(other, ComplexFilter): 

111 return NotImplemented 

112 return ( 

113 self.uid == other.uid 

114 and self.relationship == other.relationship 

115 and self.filter_by_dict == other.filter_by_dict 

116 ) 

117 

118 @override 

119 def __hash__(self) -> int: 

120 return hash( 

121 ( 

122 self.uid, 

123 self.relationship, 

124 frozenset(self.filter_by_dict.items()) if self.filter_by_dict else None, 

125 ), 

126 )