Method get_sql_engine

dataherald/sql_database/base.py:69–113

(
        cls, database_info: DatabaseConnection, refresh_connection=False
    )

    @classmethod
    def get_sql_engine(
        cls, database_info: DatabaseConnection, refresh_connection=False
    ) -> "SQLDatabase":
        logger.info(f"Connecting db: {database_info.id}")
        try:
            if (
                database_info.id
                and database_info.id in DBConnections.db_connections
                and not refresh_connection
            ):
                sql_database = DBConnections.db_connections[database_info.id]
                sql_database.engine.connect()
                return sql_database
        except OperationalError:
            pass

        fernet_encrypt = FernetEncrypt()
        try:
            if database_info.use_ssh:
                engine = cls.from_uri_ssh(database_info)
                DBConnections.add(database_info.id, engine)
                return engine
        except Exception as e:
            raise SSHInvalidDatabaseConnectionError(
                "Invalid SSH connection", description=str(e)
            ) from e
        try:
            db_uri = unquote(fernet_encrypt.decrypt(database_info.connection_uri))

            file_path = database_info.path_to_credentials_file
            if file_path and file_path.lower().startswith("s3"):
                s3 = S3()
                file_path = s3.download(file_path)

            if db_uri.lower().startswith("bigquery"):
                db_uri = db_uri + f"?credentials_path={file_path}"

            engine = cls.from_uri(db_uri)
            engine.engine.connect()
            DBConnections.add(database_info.id, engine)
        except Exception as e:
            raise InvalidDBConnectionError(  # noqa: B904
                f"Unable to connect to db: {database_info.alias}", description=str(e)
            )
        return engine

    @classmethod
    def extract_parameters(cls, input_string):