refreshnow method

[0xjjoyy]

Please consider to add the a public method to allow the ability to refresh a secret on demand. Similar to how the Java cache library has a refreshNow() method.

Currently the python methods are 'internal' class methods (though still accessible).

I can do the following to invoke a refresh, though would be more straightforward if it was supported by the class method.

            secret_cache_item._refresh_needed=True
            secret_cache_item._execute_refresh()

[0xjjoyy]

This is my get_new_connection method for illustration.

    def get_new_connection(self, conn_params):
        try:
            logger.info("get connection")
            self.get_conn_params_from_secrets_manager(conn_params)
            conn =super(DatabaseWrapper,self).get_new_connection(conn_params)
            return conn
        except MySQLdb.OperationalError as e:
            error_code=e.args[0]
            if error_code!=1045:
                raise e

            logger.info("Authentication error. Going to refresh secret and try again.")
            secret_cache_item=self.cache_secrets_manager._get_cached_secret('test/appbeta/mysql')
            secret_cache_item._refresh_needed=True
            secret_cache_item._execute_refresh()
            self.get_conn_params_from_secrets_manager(conn_params) 
            conn =super(DatabaseWrapper,self).get_new_connection(conn_params)
            return conn

[tramitws]

This is my get_new_connection method for illustration.

    def get_new_connection(self, conn_params):
        try:
            logger.info("get connection")
            self.get_conn_params_from_secrets_manager(conn_params)
            conn =super(DatabaseWrapper,self).get_new_connection(conn_params)
            return conn
        except MySQLdb.OperationalError as e:
            error_code=e.args[0]
            if error_code!=1045:
                raise e

            logger.info("Authentication error. Going to refresh secret and try again.")
            secret_cache_item=self.cache_secrets_manager._get_cached_secret('test/appbeta/mysql')
            secret_cache_item._refresh_needed=True
            secret_cache_item._execute_refresh()
            self.get_conn_params_from_secrets_manager(conn_params) 
            conn =super(DatabaseWrapper,self).get_new_connection(conn_params)
            return conn

@0xjjoyy
Thanks for that
Does it work for you allow you to get the latest secret after rotate ?
I tried to do the same with this code below and it doesn't bring the latest secret after rotate

`

client = botocore.session.get_session().create_client('secretsmanager', region_name=region_name)
cache_secrets_manager = SecretCache(config=SecretCacheConfig(), client=client)
secret_cache_item=cache_secrets_manager._get_cached_secret(secret_name)
secret_cache_item._refresh_needed = True
secret_cache_item._execute_refresh()
db_secret = json.loads(cache_secrets_manager.get_secret_string(secret_name))

`

[tramitws]

Also doing this custom db doesn't do the work

import logging

import botocore
import botocore.session
from aws_secretsmanager_caching import SecretCache, SecretCacheConfig

from django.db.backends.postgresql import base

import json

logger = logging.getLogger(__name__)


class DatabaseCredentials:
    def __init__(self):
        logger.info("init secrets manager database credentials")
        client = botocore.session.get_session().create_client('secretsmanager')
        cache_config = SecretCacheConfig()
        self.cache_secrets_manager = SecretCache(config=cache_config, client=client)
        self.secret_id = "pg-prod-cred"

    def get_conn_params_from_secrets_manager(self):
        secret_json = self.cache_secrets_manager.get_secret_string(self.secret_id)
        secret_dict = json.loads(secret_json)
        return secret_dict

    def refresh_now(self):
        secret_cache_item = self.cache_secrets_manager._get_cached_secret(self.secret_id)
        secret_cache_item._refresh_needed = True
        secret_cache_item._execute_refresh()
        print("refresh_now")


databasecredentials = DatabaseCredentials()


class DatabaseWrapper(base.DatabaseWrapper):
    def get_new_connection(self, conn_params):
        try:
            logger.info("get connection")
            databasecredentials.get_conn_params_from_secrets_manager()
            conn = super(DatabaseWrapper, self).get_new_connection(conn_params)
            return conn
        except DatabaseWrapper.Database.OperationalError as e:
            error_code = e.args[0]
            if error_code != 1045:
                raise e

            logger.info("Authentication error. Going to refresh secret and try again.")
            databasecredentials.refresh_now()
            conn_params = databasecredentials.get_conn_params_from_secrets_manager()
            conn = super(DatabaseWrapper, self).get_new_connection(conn_params)
            logger.info("Successfully refreshed secret and established new database connection.")
            return conn