Coverage for filip/utils/filter.py: 96%

1""" 

2Filter functions to keep client code clean and easy to use. 

3""" 

4 

5from typing import List, Union 

6from filip.clients.ngsi_v2 import ContextBrokerClient 

7from filip.models import FiwareHeader 

8from filip.models.ngsi_v2.iot import Device, ServiceGroup 

9from filip.models.ngsi_v2.subscriptions import Subscription 

10from requests.exceptions import RequestException 

11 

12 

13def filter_device_list( 

14 devices: List[Device], 

15 device_ids: Union[str, List[str]] = None, 

16 entity_names: Union[str, List[str]] = None, 

17 entity_types: Union[str, List[str]] = None, 

18) -> List[Device]: 

19 """ 

20 Filter the given device list based on conditions 

21 

22 Args: 

23 devices: device list that need to be filtered 

24 device_ids: A list of device_id as filter condition 

25 entity_names: A list of entity_name (e.g. entity_id) as filter condition. 

26 entity_types: A list of entity_type as filter condition 

27 

28 Returns: 

29 List of matching devices 

30 """ 

31 if device_ids: 

32 if isinstance(device_ids, (list, str)): 

33 if isinstance(device_ids, str): 

34 device_ids = [device_ids] 

35 devices = [device for device in devices if device.device_id in device_ids] 

36 else: 

37 raise TypeError("device_ids must be a string or a list of strings!") 

38 

39 if entity_names: 

40 if isinstance(entity_names, (list, str)): 

41 if isinstance(entity_names, str): 

42 entity_names = [entity_names] 

43 devices = [ 

44 device for device in devices if device.entity_name in entity_names 

45 ] 

46 else: 

47 raise TypeError("entity_names must be a string or a list of strings!") 

48 

49 if entity_types: 

50 if isinstance(entity_types, (list, str)): 

51 if isinstance(entity_types, str): 

52 entity_types = [entity_types] 

53 devices = [ 

54 device for device in devices if device.entity_type in entity_types 

55 ] 

56 else: 

57 raise TypeError("entity_types must be a string or a list of strings!") 

58 

59 return devices 

60 

61 

62def filter_subscriptions_by_entity( 

63 entity_id: str, 

64 entity_type: str, 

65 url: str = None, 

66 fiware_header: FiwareHeader = None, 

67 subscriptions: List[Subscription] = None, 

68) -> List[Subscription]: 

69 """ 

70 Function that filters subscriptions based on the entity id or id pattern 

71 and entity type or type pattern. The function can be used in two ways, 

72 wither pass list of subscriptions to filter based on entity or directly pass 

73 client information to filter subscriptions in a single request. 

74 

75 Args: 

76 entity_id: Id of the entity to be matched 

77 entity_type: Type of the entity to be matched 

78 url: Url of the context broker service 

79 fiware_header: Fiware header of the tenant 

80 subscriptions: List of subscriptions to filter 

81 Returns: 

82 list of subscriptions by entity 

83 """ 

84 if not subscriptions: 

85 client = ContextBrokerClient(url=url, fiware_header=fiware_header) 

86 subscriptions = client.get_subscription_list() 

87 filtered_subscriptions = [] 

88 for subscription in subscriptions: 

89 for entity in subscription.subject.entities: 

90 if entity.id == entity_id or ( 

91 entity.idPattern is not None and entity.idPattern.match(entity_id) 

92 ): 

93 if entity.type == entity_type or ( 

94 entity.typePattern is not None 

95 and entity.typePattern.match(entity_type) 

96 ): 

97 filtered_subscriptions.append(subscription) 

98 return filtered_subscriptions 

99 

100 

101def filter_group_list( 

102 group_list: List[ServiceGroup], 

103 resources: Union[str, List[str]] = None, 

104 apikeys: Union[str, List[str]] = None, 

105) -> List[ServiceGroup]: 

106 """ 

107 Filter service group based on resource and apikey. 

108 

109 Args: 

110 group_list: The list of service groups that need to be filtered 

111 resources: see ServiceGroup model 

112 apikeys: see ServiceGroup 

113 

114 Returns: a single service group or Not Found Error 

115 """ 

116 if resources: 

117 if isinstance(resources, (list, str)): 

118 if isinstance(resources, str): 

119 resources = [resources] 

120 group_list = [group for group in group_list if group.resource in resources] 

121 else: 

122 raise TypeError("resources must be a string or a list of strings!") 

123 

124 if apikeys: 

125 if isinstance(apikeys, (list, str)): 

126 if isinstance(apikeys, str): 

127 apikeys = [apikeys] 

128 group_list = [group for group in group_list if group.apikey in apikeys] 

129 else: 

130 raise TypeError("apikeys must be a string or a list of strings!") 

131 

132 return group_list