diff --git a/database_test_python/db_config.py b/database_test_python/db_config.py new file mode 100644 index 0000000..baa600b --- /dev/null +++ b/database_test_python/db_config.py @@ -0,0 +1,63 @@ +import os +from getpass import getpass + +# Global cache for database passwords to avoid asking multiple times for the same database +_password_cache = {} + +def get_db_password(server_name: str, db_name: str) -> str: + conn_name = f"{server_name}.{db_name}" + + # Only ask once per target database per script run + if conn_name in _password_cache: + return _password_cache[conn_name] + + # Hardcoded fallback per user request + if server_name == "109portal": + password = "YourActualStrongPasswordForPostgres!" + elif server_name == "80portal": + password = "H3IkLUt8K!x" + else: + env_var_name = f"{server_name.upper()}_PG_PASSWORD" + password = os.getenv(env_var_name) + + if not password: + # Prompt specific to 'name'.'database' + password = getpass(f"{conn_name} password: ") + + _password_cache[conn_name] = password + return password + +def get_pg_conn_kwargs(prefix: str) -> dict: + if prefix == "PORTAL": + server_name = "80portal" + default_port = 15434 + default_db = "log_platform" + elif prefix == "TEST" or prefix == "TARGET": + server_name = "109portal" + default_port = 15433 + default_db = "test" + else: # SOURCE + server_name = "80portal" + default_port = 15434 + default_db = "log_platform" + + dbname = os.getenv(f"{prefix}_PG_DB", default_db) + + return { + "host": os.getenv(f"{prefix}_PG_HOST", "blv-rd.tech"), + "port": int(os.getenv(f"{prefix}_PG_PORT", default_port)), + "dbname": dbname, + "user": os.getenv(f"{prefix}_PG_USER", "log_admin"), + "password": get_db_password(server_name, dbname) + } + +def get_db_url(prefix: str) -> str: + kwargs = get_pg_conn_kwargs(prefix) + return f"postgresql+psycopg2://{kwargs['user']}:{kwargs['password']}@{kwargs['host']}:{kwargs['port']}/{kwargs['dbname']}" + +DB_CONFIG = { + "SOURCE": lambda: get_pg_conn_kwargs("SOURCE"), + "TARGET": lambda: get_pg_conn_kwargs("TARGET"), + "PORTAL": lambda: get_pg_conn_kwargs("PORTAL"), + "TEST": lambda: get_pg_conn_kwargs("TEST"), +}