Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/c7n/registry.py: 92%

1# Copyright The Cloud Custodian Authors. 

2# SPDX-License-Identifier: Apache-2.0 

3 

4 

5class PluginRegistry: 

6 """A plugin registry 

7 

8 Custodian is intended to be innately pluggable both internally and 

9 externally, for resource types and their filters and actions. 

10 

11 This plugin registry abstraction provides the core mechanism for 

12 that. Its a simple string to class map, with python package 

13 entry_point loading for external plugins. 

14 

15 As an example of defining an external plugin using a python package 

16 

17 .. code-block:: python 

18 

19 setup( 

20 name="custodian_cmdb", 

21 description="Custodian filters for interacting with internal CMDB" 

22 version='1.0', 

23 packages=find_packages(), 

24 entry_points={ 

25 'console_scripts': [ 

26 'custodian.ec2.filters = custodian_cmdb:filter_ec2']}, 

27 ) 

28 

29 For loading the plugins we can simply invoke method:load_plugins like 

30 so:: 

31 

32 PluginRegistry('ec2.filters').load_plugins() 

33 

34 """ 

35 

36 EVENT_REGISTER = 0 

37 EVENT_FINAL = 1 

38 EVENTS = (EVENT_REGISTER, EVENT_FINAL) 

39 

40 def __init__(self, plugin_type): 

41 self.plugin_type = plugin_type 

42 self._factories = {} 

43 self._subscribers = [] 

44 

45 def subscribe(self, func): 

46 self._subscribers.append(func) 

47 

48 def register(self, name, klass=None, condition=True, 

49 condition_message="Missing dependency for {}", 

50 aliases=None): 

51 if not condition and klass: 

52 return klass 

53 # invoked as function 

54 if klass: 

55 klass.type = name 

56 klass.type_aliases = aliases 

57 self._factories[name] = klass 

58 return klass 

59 

60 # invoked as class decorator 

61 def _register_class(klass): 

62 if not condition: 

63 return klass 

64 self._factories[name] = klass 

65 klass.type = name 

66 klass.type_aliases = aliases 

67 return klass 

68 return _register_class 

69 

70 def unregister(self, name): 

71 if name in self._factories: 

72 del self._factories[name] 

73 

74 def notify(self, key=None): 

75 for subscriber in self._subscribers: 

76 subscriber(self, key) 

77 

78 def __contains__(self, key): 

79 return key in self._factories 

80 

81 def __getitem__(self, name): 

82 v = self.get(name) 

83 if v is None: 

84 raise KeyError(name) 

85 return v 

86 

87 def __len__(self): 

88 return len(self._factories) 

89 

90 def get(self, name): 

91 factory = self._factories.get(name) 

92 

93 if factory: 

94 return factory 

95 

96 return next((v for k, v in self._factories.items() 

97 if v.type_aliases and name in v.type_aliases), 

98 None) 

99 

100 def keys(self): 

101 return self._factories.keys() 

102 

103 def values(self): 

104 return self._factories.values() 

105 

106 def items(self): 

107 return self._factories.items()