import uuid from cachelib.simple import SimpleCache from ddtrace import tracer from secrets_manager.python_ext import PythonSecretsManager from grass import config from grass.api import logger from grass.logger import CustomAdapter from grass.utils.jwt_enabled_services_cache_refresh import refreshCache class jwtEnabledServices: __instance = None simpleInstance = None refreshCache = None @classmethod def getInstance(cls): if not cls.__instance: cls.__instance = jwtEnabledServices() cls.simpleInstance = SimpleCache() cls.refreshCache = refreshCache() return cls.__instance @tracer.wrap() def getJwtEnabledServices(self): try: secrets_manager_client = self.get_secret_manager_client() secrets = secrets_manager_client.get_cred(config.JWT_SECRET_NAME) return secrets except Exception as e: log = self.logger() log.info(f'JwtEnabledServices: ' f'Exception : {e}') raise e @tracer.wrap() def cacheJwtEnabledServices(self, isThread=False): secrets = self.getJwtEnabledServices() self.setCache(config.JWT_SECRET_NAME, secrets, 0) if not isThread: log = self.logger() log.info('JwtEnabledServices: ' 'Start background thread to refresh cache') self.refreshCache.start() def deleteCache(self, key): self.simpleInstance.delete(key) def setCache(self, key, secrets, ttl): self.simpleInstance.set(key, secrets, ttl) def isCached(self, key): return self.simpleInstance.has(key) def getCache(self, key): return self.simpleInstance.get(key) def get_secret_manager_client(self): return PythonSecretsManager( application_context=False, environment=config.environment, service_name=config.JWT_SERVICE_NAME, ) def logger(self): correlation_id = str(uuid.uuid1()) context = dict(correlation_id=correlation_id) return CustomAdapter(logger, context)