def execute_statement()

in lambdas/helper/dal.py [0:0]

• 24 lines of code
• 4 McCabe index (conditional complexity)

    def execute_statement(self, sql_stmt, sql_params=[], transaction_id=None):
        parameters = f' with parameters: {sql_params}' if len(sql_params) > 0 else ''
        logger.debug(f'Running SQL statement: {sql_stmt}{parameters}')
        DataAccessLayer._xray_start('execute_statement')
        try:
            DataAccessLayer._xray_add_metadata('sql_statement', sql_stmt)
            parameters = {
                'secretArn': self._db_credentials_secrets_store_arn,
                'database': self._database_name,
                'resourceArn': self._db_cluster_arn,
                'sql': sql_stmt,
                'parameters': sql_params
            }
            if transaction_id is not None:
                parameters['transactionId'] = transaction_id
            result = self._rdsdata_client.execute_statement(**parameters)
        except Exception as e:
            logger.debug(f'Error running SQL statement (error class: {e.__class__})')
            raise DataAccessLayerException(e) from e
        else:
            DataAccessLayer._xray_add_metadata('rdsdata_executesql_result', json.dumps(result))
            return result
        finally:
           DataAccessLayer._xray_stop()