To Nha Notes | May 16, 2023, 9:12 a.m.
Below are shared common module for feature flag with works on Airflow environment.
----------------- path/to/common/featureflag.py ------------------
import logging
from typing import Any, List
import requests
from airflow.models import Variable
logger = logging.getLogger(__name__)
FEATURE_FLAG_SERVER_URL_VAR = "feature_flag_server_url"
FEATURE_FLAG_SEGMENT_KEY_VAR = "feature_flag_segment_key_{flag_key}"
class FeatureFlagSegmentConstraintOperators:
EQUAL = "eq"
NOT_EQUAL = "neq"
class FeatureFlagKeys:
SITE_VISITED_PATHS_AUTO_SUGGESTION_ENABLED = "site_visited_paths_auto_suggestion_enabled"
class FeatureFlag:
def get_segment(self, flag_key: str) -> Any:
"""
Request to self-host feature flag server Flipt to get segment information.
API document: https://www.flipt.io/docs/reference/segments/get-segment
flag_key: defined in FeatureFlagKeys used to get segment key in Airflow variable.
return: segment JSON as below example
{
"key": "<SEGMENT_KEY>",
"name": "<SEGMENT_NAME>",
"description": "",
"createdAt": "2023-05-15T10:04:59Z",
"updatedAt": "2023-05-15T10:04:59Z",
"constraints": [
{
"id": "4fe539f2-7536-4ad9-86ca-d6b965369905",
"segmentKey": "<SEGMENT_KEY>",
"type": "STRING_COMPARISON_TYPE",
"property": "organization_id",
"operator": "eq",
"value": "00000000000040008000000000000000",
"createdAt": "2023-05-15T10:07:14Z",
"updatedAt": "2023-05-15T10:07:14Z"
},
...
]
}
"""
feature_flag_server_url: str = Variable.get(FEATURE_FLAG_SERVER_URL_VAR, default_var="")
if not feature_flag_server_url:
logger.info(f"The Airflow variable \"{FEATURE_FLAG_SERVER_URL_VAR}\" for feature flag is not defined yet.")
return
segment_key_var_name = FEATURE_FLAG_SEGMENT_KEY_VAR.format(**{"flag_key": flag_key})
segment_key: str = Variable.get(segment_key_var_name, default_var="")
if not segment_key:
logger.info(f"The Airflow variable \"{segment_key_var_name}\" for feature flag is not defined yet.")
return
segment_api_url = f"{feature_flag_server_url}/api/v1/segments/{segment_key}"
logger.info(f"Request to API \"{segment_api_url}\" to get feature flag segment")
response = requests.get(segment_api_url)
response.raise_for_status()
segment = response.json()
return segment
def get_segment_constraint_values(
self,
flag_key: str,
constraint_property: str,
constraint_operator: str = FeatureFlagSegmentConstraintOperators.EQUAL,
) -> List[str]:
constraint_values: List[str] = []
segment = self.get_segment(flag_key)
if not segment:
return constraint_values
constraints = segment.get("constraints", [])
if not constraints:
return constraint_values
for c in constraints:
if c["property"] == constraint_property and c["operator"] == constraint_operator:
constraint_values.append(c["value"])
return constraint_values