nofusscomputing/centurion_erp
1
0
Fork 0
Code Issues Wiki Activity

feat(devops): Add Feature Flagging lib

Browse Source 
ref: #675 #575
This commit is contained in:
Jon Lockwood
2025-03-13 19:47:44 +09:30
parent 1d9580f14b
commit c5107861a1
4 changed files with 433 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
app/centurion_feature_flag/lib/__init__.py Normal file
View File

315
app/centurion_feature_flag/lib/feature_flag.py Normal file
View File

@ -0,0 +1,315 @@
import json
import requests
from datetime import datetime
from dateutil.parser import parse
from pathlib import Path
from centurion_feature_flag.lib.serializer import FeatureFlag
class CenturionFeatureFlagging:
"""Centurion ERP Feature Flags
This class contains all required methods so as to use feature flags
provided by a Centurion ERP deployment.
Examples:
Checking if feature flagging is usable can be done with:
>>> ff = CenturionFeatureFlagging(
>>> 'http://127.0.0.1:8002/api/v2/public/1/flags/2844',
>>> 'Centurion ERP',
>>> './your-cache-dir'
>>> )
>>> if ff:
>>> print('ok')
ok
To use a feature flag, in this case `2025-00007` can be achived with:
>>> if ff["2025-00007"]:
>>> print('ok')
ok
Note: This assumes that feature flag `2025-00007` is enabled. If it is not
`false` will be returned as the boolean check returns the flags `enabled`
value.
Args:
url (str): URL of the Centurion Instance to query
user_agent (str): User Agent to report to Centurion Instance this
should be the name of your application
cache_dir (str): Directory where the feature flag cache file is saved.
unique_id (str, optional): Unique ID of the application that is
reporting to Centurion ERP
version (str, optional): The version of your application
Attributes:
__len__ (int): Count of feature flags
__bool__ (bool): Feature Flag fetch was successful
CenturionFeatureFlagging[<feature flag>] (dict): Feature flag data
get (None): Make a http request to the Centurion ERP
instance.
"""
_cache_date: datetime = None
"""Date the feature flag file was last saved"""
_cache_dir: str = None
_feature_flags: list = None
_feature_flag_filename: str = 'feature_flags.json'
_headers: dict = {
"Accept": "application/json",
}
_last_modified: datetime = None
_response: requests.Response = None
_ssl_verify: bool = True
_url: str = None
def __init__(
self,
url,
user_agent,
cache_dir: str,
unique_id = None,
version = None,
):
if not str(cache_dir).endswith('/'):
raise AttributeError(f'cache directory {cache_dir} must end with trailing slash `/`')
self._url = url
self._cache_dir = cache_dir
if version is not None:
self._headers.update({
'User-Agent': f'{user_agent} 0.0'
})
else:
self._headers.update({
'User-Agent': f'{user_agent} {version}'
})
if unique_id is not None:
self._headers.update({
'client-id': unique_id
})
def __bool__(self) -> bool:
if(
(
(
getattr(self._response, 'status_code', 0) == 200
or getattr(self._response, 'status_code', 0) == 304
)
and self._feature_flags is not None
)
or ( # Feature flags were loaded from file
self._feature_flags is not None
and self._last_modified is not None
)
):
return True
return False
def __getitem__(self, key: str, raise_exceptions: bool = False) -> dict:
""" Fetch a Feature Flag
Args:
key (str): Feature Flag id to fetch.
raise_exceptions (bool, optional): Raise an exception if the key is
not found. Default `False`
Raises:
KeyError: The specified Feature Flag does not exist. Only if arg `raise_exceptions=True`
Returns:
dict: A complete Feature Flag.
"""
if self._feature_flags is None:
print('Feature Flagging has not been completly initialized.')
print(' please ensure that the feature flags have been downloaded.')
return False
if(
self._feature_flags.get(key, None) is None
and raise_exceptions
):
raise KeyError(f'Feature Flag "{key}" does not exist')
elif(
not raise_exceptions
and self._feature_flags.get(key, None) is None
):
return False
return self._feature_flags[key]
def __len__(self) -> int:
"""Count the Feature Flags
Returns:
int: Total number of feature flags.
"""
return len(self._feature_flags)
def get( self ):
""" Get the available Feature Flags
Will first check the filesystem for file `feature_flags.json` and if
the file is '< 4 hours' old, will load the feature flags from the file.
If the file does not exist or the file is '> 4 hours' old, the feature
flags will be fetched from Centurion ERP.
"""
url = self._url
fetched_flags: list = []
feature_flag_path = self._cache_dir + self._feature_flag_filename
feature_flag_file = Path(feature_flag_path)
if feature_flag_file.is_file():
if feature_flag_file.lstat().st_mtime > datetime.now().timestamp() - (4 * 3580): # -20 second buffer
# Only open file if less than 4 hours old
with open(feature_flag_path, 'r') as saved_feature_flags:
fetched_flags = json.loads(saved_feature_flags.read())
self._cache_date = datetime.fromtimestamp(feature_flag_file.lstat().st_mtime)
url = None
response = None
while(url is not None):
try:
resp = requests.get(
headers = self._headers,
timeout = 3,
url = url,
verify = self._ssl_verify,
)
if response is None: # Only save first request
response = resp
self._response = response
fetched_flags += resp.json()['results']
if resp.status_code == 304: # Nothing has changed, exit the loop
url = None
else: # Fetch next page of results
url = resp.json()['next']
except requests.exceptions.ConnectionError as err:
url = None
if(
getattr(response, 'status_code', 0) == 200
or len(fetched_flags) > 0
):
feature_flags: dict = {}
for entry in fetched_flags:
[*key], [*flag] = zip(*entry.items())
feature_flags.update({
key[0]: FeatureFlag(key[0], flag[0])
})
self._feature_flags = feature_flags
if response is not None:
if response.headers.get('last-modified', None) is not None:
self._last_modified = datetime.strptime(response.headers['last-modified'], '%a, %d %b %Y %H:%M:%S %z')
else:
last_mod_date: datetime = datetime.fromtimestamp(0)
for item in self._feature_flags:
parsed_date = parse(self._feature_flags[item].modified)
if parsed_date.timestamp() > last_mod_date.timestamp():
last_mod_date = parsed_date
self._last_modified = last_mod_date
if getattr(response, 'status_code', 0) == 200:
with open(feature_flag_path, 'w') as feature_flag_file:
feature_flag_file.write(self.toJson())
self._cache_date = datetime.now()
def toJson(self):
obj = []
for entry in self._feature_flags:
obj += [
self._feature_flags[entry].dump()
]
return json.dumps(obj)

117
app/centurion_feature_flag/lib/serializer.py Normal file
View File

@ -0,0 +1,117 @@
from datetime import datetime
class FeatureFlag:
"""Centurion ERP Feature Flag
Contains a Centurion ERP feature flag.
Args:
key (str):
Attributes:
__bool__ (bool): Enabled value
__str__ (str): Name of the feature flag
key (str): Feature Flag key
name (str): Feature Flag name
description (str): Feature Flag Description
enabled (bool): Enabled value of the feature flag
created (datetime): Creation date of the feature flag
modified (datetime): Date when feature flag was last modified
"""
_key: str = None
_name: str = None
_description: str = None
_enabled: bool = None
_created: datetime = None
_modified: datetime = None
def __init__(self, key, flag: dict):
self._key = key
self._name = flag['name']
self._description = flag['description']
self._enabled = flag['enabled']
self._created = flag['created']
self._modified = flag['modified']
def __bool__(self) -> bool:
"""Feature Flag Enabled
Returns:
bool: Feature flag enabled value.
"""
return self._enabled
def __str__(self) -> str:
"""Fetch name of Feature Flag
Returns:
str: Name of the Feature Flag
"""
return self._name
@property
def key(self) -> str:
return self._key
@property
def name(self) -> str:
return self._name
@property
def description(self) -> str:
return self._description
@property
def enabled(self) -> bool:
return self._enabled
@property
def created(self) -> datetime:
return self._created
@property
def modified(self) -> datetime:
return self._modified
def dump(self) -> dict:
return {
self.key: {
'name': self.name,
'description': self.description,
'enabled': self.enabled,
'created': self.created,
'modified': self.modified
}
}