in evalbench/databases/db.py [0:0]
def __init__(self, db_config):
self.db_path = db_config["database_path"]
self.db_name = db_config["database_name"]
self.db_type = db_config["db_type"]
self.username = db_config.get("user_name") or ""
if self.db_type == "sqlite":
self.extension = db_config.get("extension") or ".db"
if "password" in db_config and db_config["password"]:
self.password = db_config["password"]
elif "secret_manager_path" in db_config and db_config["secret_manager_path"]:
self.password = get_db_secret(db_config["secret_manager_path"])
else:
# no password
self.password = None
# Setup the concurrency requirements
self.execs_per_minute = db_config["max_executions_per_minute"]
self.max_attempts = 3
self.semaphore = Semaphore(self.execs_per_minute)
# Maintain setup / teardown information
self.tmp_dbs = []
self.tmp_users = []
self.was_re_setup_this_session = False
# Initialize the Redis cache client
self.cache_client = get_cache_client(db_config)