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

1""" 

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

3""" 

4from typing import List, Union 

5from filip.clients.ngsi_v2 import ContextBrokerClient 

6from filip.models import FiwareHeader 

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

8from filip.models.ngsi_v2.subscriptions import Subscription 

9from requests.exceptions import RequestException 

10 

11 

12def filter_device_list(devices: List[Device], 

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

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

15 entity_types: Union[str, List[str]] = None) -> List[Device]: 

16 """ 

17 Filter the given device list based on conditions 

18 

19 Args: 

20 devices: device list that need to be filtered 

21 device_ids: A list of device_id as filter condition 

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

23 entity_types: A list of entity_type as filter condition 

24 

25 Returns: 

26 List of matching devices 

27 """ 

28 if device_ids: 

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

30 if isinstance(device_ids, str): 

31 device_ids = [device_ids] 

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

33 else: 

34 raise TypeError('device_ids must be a string or a list of strings!') 

35 

36 if entity_names: 

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

38 if isinstance(entity_names, str): 

39 entity_names = [entity_names] 

40 devices = [device for device in devices if device.entity_name in entity_names] 

41 else: 

42 raise TypeError('entity_names must be a string or a list of strings!') 

43 

44 if entity_types: 

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

46 if isinstance(entity_types, str): 

47 entity_types = [entity_types] 

48 devices = [device for device in devices if device.entity_type in entity_types] 

49 else: 

50 raise TypeError('entity_types must be a string or a list of strings!') 

51 

52 return devices 

53 

54 

55def filter_subscriptions_by_entity(entity_id: str, 

56 entity_type: str, 

57 url: str = None, 

58 fiware_header: FiwareHeader = None, 

59 subscriptions: List[Subscription] = None, 

60 ) -> List[Subscription]: 

61 """ 

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

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

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

65 client information to filter subscriptions in a single request. 

66 

67 Args: 

68 entity_id: Id of the entity to be matched 

69 entity_type: Type of the entity to be matched 

70 url: Url of the context broker service 

71 fiware_header: Fiware header of the tenant 

72 subscriptions: List of subscriptions to filter 

73 Returns: 

74 list of subscriptions by entity 

75 """ 

76 if not subscriptions: 

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

78 subscriptions = client.get_subscription_list() 

79 filtered_subscriptions = [] 

80 for subscription in subscriptions: 

81 for entity in subscription.subject.entities: 

82 if entity.id == entity_id or ( 

83 entity.idPattern is not None 

84 and entity.idPattern.match(entity_id)): 

85 if entity.type == entity_type or \ 

86 (entity.typePattern is not None and 

87 entity.typePattern.match(entity_type)): 

88 filtered_subscriptions.append(subscription) 

89 return filtered_subscriptions 

90 

91 

92def filter_group_list(group_list: List[ServiceGroup], 

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

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

95 ) -> List[ServiceGroup]: 

96 """ 

97 Filter service group based on resource and apikey. 

98 

99 Args: 

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

101 resources: see ServiceGroup model 

102 apikeys: see ServiceGroup 

103 

104 Returns: a single service group or Not Found Error 

105 """ 

106 if resources: 

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

108 if isinstance(resources, str): 

109 resources = [resources] 

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

111 else: 

112 raise TypeError('resources must be a string or a list of strings!') 

113 

114 if apikeys: 

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

116 if isinstance(apikeys, str): 

117 apikeys = [apikeys] 

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

119 else: 

120 raise TypeError('apikeys must be a string or a list of strings!') 

121 

122 return group_list 

123 

124