1
1
import logging
2
- from typing import Any , Dict , List , Optional
2
+ from typing import Any , Dict , List , Optional , cast
3
3
4
4
from . import schema
5
5
from .exceptions import ConfigurationError
@@ -12,8 +12,10 @@ class ConfigurationStore:
12
12
def __init__ (self , schema_fetcher : SchemaFetcher ):
13
13
"""constructor
14
14
15
- Args:
16
- schema_fetcher (SchemaFetcher): A schema JSON fetcher, can be AWS AppConfig, Hashicorp Consul etc.
15
+ Parameters
16
+ ----------
17
+ schema_fetcher: SchemaFetcher
18
+ A schema JSON fetcher, can be AWS AppConfig, Hashicorp Consul etc.
17
19
"""
18
20
self ._logger = logger
19
21
self ._schema_fetcher = schema_fetcher
@@ -39,12 +41,12 @@ def _match_by_action(self, action: str, condition_value: Any, context_value: Any
39
41
def _is_rule_matched (self , feature_name : str , rule : Dict [str , Any ], rules_context : Dict [str , Any ]) -> bool :
40
42
rule_name = rule .get (schema .RULE_NAME_KEY , "" )
41
43
rule_default_value = rule .get (schema .RULE_DEFAULT_VALUE )
42
- conditions : Dict [ str , str ] = rule .get (schema .CONDITIONS_KEY )
44
+ conditions = cast ( List [ Dict ], rule .get (schema .CONDITIONS_KEY ) )
43
45
44
46
for condition in conditions :
45
- context_value = rules_context .get (condition .get (schema .CONDITION_KEY ))
47
+ context_value = rules_context .get (str ( condition .get (schema .CONDITION_KEY ) ))
46
48
if not self ._match_by_action (
47
- condition .get (schema .CONDITION_ACTION ),
49
+ condition .get (schema .CONDITION_ACTION , "" ),
48
50
condition .get (schema .CONDITION_VALUE ),
49
51
context_value ,
50
52
):
@@ -73,7 +75,7 @@ def _handle_rules(
73
75
for rule in rules :
74
76
rule_default_value = rule .get (schema .RULE_DEFAULT_VALUE )
75
77
if self ._is_rule_matched (feature_name , rule , rules_context ):
76
- return rule_default_value
78
+ return bool ( rule_default_value )
77
79
# no rule matched, return default value of feature
78
80
logger .debug (
79
81
f"no rule matched, returning default value of feature, feature_default_value={ feature_default_value } , "
@@ -95,20 +97,18 @@ def get_configuration(self) -> Dict[str, Any]:
95
97
Dict[str, Any]
96
98
parsed JSON dictionary
97
99
"""
98
- config : Dict [
99
- str , Any
100
- ] = (
101
- self ._schema_fetcher .get_json_configuration ()
102
- ) # parse result conf as JSON, keep in cache for self.max_age seconds
100
+ # parse result conf as JSON, keep in cache for self.max_age seconds
101
+ config = self ._schema_fetcher .get_json_configuration ()
103
102
# validate schema
104
103
self ._schema_validator .validate_json_schema (config )
105
104
return config
106
105
107
106
def get_feature_toggle (
108
107
self , * , feature_name : str , rules_context : Optional [Dict [str , Any ]] = None , value_if_missing : bool
109
108
) -> bool :
110
- """get a feature toggle boolean value. Value is calculated according to a set of rules and conditions.
111
- see below for explanation.
109
+ """Get a feature toggle boolean value. Value is calculated according to a set of rules and conditions.
110
+
111
+ See below for explanation.
112
112
113
113
Parameters
114
114
----------
@@ -123,7 +123,7 @@ def get_feature_toggle(
123
123
configuration from appconfig
124
124
125
125
Returns
126
- ----------
126
+ ------
127
127
bool
128
128
calculated feature toggle value. several possibilities:
129
129
1. if the feature doesn't appear in the schema or there has been an error fetching the
@@ -157,16 +157,16 @@ def get_feature_toggle(
157
157
f"no rules found, returning feature default value, feature_name={ feature_name } , "
158
158
f"default_value={ feature_default_value } "
159
159
)
160
- return feature_default_value
160
+ return bool ( feature_default_value )
161
161
# look for first rule match
162
162
logger .debug (
163
163
f"looking for rule match, feature_name={ feature_name } , feature_default_value={ feature_default_value } "
164
164
)
165
165
return self ._handle_rules (
166
166
feature_name = feature_name ,
167
167
rules_context = rules_context ,
168
- feature_default_value = feature_default_value ,
169
- rules = rules_list ,
168
+ feature_default_value = bool ( feature_default_value ) ,
169
+ rules = cast ( List , rules_list ) ,
170
170
)
171
171
172
172
def get_all_enabled_feature_toggles (self , * , rules_context : Optional [Dict [str , Any ]] = None ) -> List [str ]:
0 commit comments