from typing import Optional from flask import _request_ctx_stack from werkzeug.local import LocalProxy from auth_api import config if config.SPLIT_API_KEY: from splitio import get_factory factory = get_factory(config.SPLIT_API_KEY, config=config.SPLIT_CONFIG) split_client, split_manager = factory.client(), factory.manager() current_user = LocalProxy(lambda: getattr(_request_ctx_stack.top, "current_user", None)) class User: def __init__(self, payload: dict, resource_group: dict = None): self.__payload: dict = payload self.__resource_group: dict = resource_group or {} self.__userinfo: dict = {} @property def user_id(self) -> Optional[str]: no_user_id = None user_id = self.__payload.get("sub", "") if "@" in user_id or not any(user_id.startswith(p) for p in config.SUB_PREFIXES): return no_user_id return user_id.split("|")[-1] @property def client_id(self) -> str: return self.__payload.get("sub", "") @property def is_m2m(self) -> bool: return not self.user_id @property def user_info(self) -> dict: # TODO: ugly validation for Auth0 support if not self.__resource_group: return self.__payload if self.__userinfo: return self.__userinfo self.__userinfo = { x.split("/")[-1]: self.__payload[x] for x in self.__payload.keys() if x.startswith(self.__resource_group) } return self.__userinfo def get_treatments(self) -> dict: if config.SPLIT_API_KEY and self.user_id: active_splits = split_manager.split_names() if active_splits: return split_client.get_treatments_with_config(self.user_id, active_splits) return {}