connection 

 1from pg8000.native import Connection
 2from pg8000.exceptions import DatabaseError
 3from extract_utils import get_secret, log_message
 4
 5
 6def connect_to_db() -> Connection:
 7    """
 8    Establishes a connection to the PostgreSQL database using credentials
 9    retrieved from AWS Secrets Manager.
10
11    This function fetches database connection details from AWS Secrets Manager,
12    and uses these details to create and return a connection object to the PostgreSQL
13    database. If an error occurs during connection, it logs the error and raises an exception.
14
15    :return: A pg8000 native.Connection object.
16    :raises DatabaseError: If there is an issue with the database connection.
17    """
18    try:
19        credentials = get_secret()
20        log_message(
21            __name__, 20, "Retrieved secrets from Secrets Manager for DB access"
22        )
23        return Connection(
24            user=credentials["USERNAME"],
25            password=credentials["PASSWORD"],
26            database=credentials["DBNAME"],
27            host=credentials["HOST"],
28            port=int(credentials["PORT"]),
29        )
30
31    except DatabaseError as e:
32        log_message(__name__, 40, f"DatabaseError: {e}")
33        raise e
def connect_to_db() -> pg8000.native.Connection: 
 7def connect_to_db() -> Connection:
 8    """
 9    Establishes a connection to the PostgreSQL database using credentials
10    retrieved from AWS Secrets Manager.
11
12    This function fetches database connection details from AWS Secrets Manager,
13    and uses these details to create and return a connection object to the PostgreSQL
14    database. If an error occurs during connection, it logs the error and raises an exception.
15
16    :return: A pg8000 native.Connection object.
17    :raises DatabaseError: If there is an issue with the database connection.
18    """
19    try:
20        credentials = get_secret()
21        log_message(
22            __name__, 20, "Retrieved secrets from Secrets Manager for DB access"
23        )
24        return Connection(
25            user=credentials["USERNAME"],
26            password=credentials["PASSWORD"],
27            database=credentials["DBNAME"],
28            host=credentials["HOST"],
29            port=int(credentials["PORT"]),
30        )
31
32    except DatabaseError as e:
33        log_message(__name__, 40, f"DatabaseError: {e}")
34        raise e

Establishes a connection to the PostgreSQL database using credentials retrieved from AWS Secrets Manager.

This function fetches database connection details from AWS Secrets Manager, and uses these details to create and return a connection object to the PostgreSQL database. If an error occurs during connection, it logs the error and raises an exception.

Returns

A pg8000 native.Connection object.

Raises
  • DatabaseError: If there is an issue with the database connection.