NVIDIA Docs Hub NVIDIA Morpheus NVIDIA Morpheus (24.10.01) morpheus.utils.sql_utils.SQLUtils

(Latest Version)

morpheus.utils.sql_utils.SQLUtils

class SQLUtils(sql_config, pool_size=5, max_overflow=5)[source]

Bases: <a href="https://docs.python.org/3/library/functions.html#object">object</a>

Utility class for working with SQL databases using SQLAlchemy.

Parameters

sql_configtyping.Dict[any, any]: Configuration parameters for the SQL connection.
pool_sizeint: The number of connections to keep in the connection pool (default: 5).
max_overflowint: The maximum number of connections that can be created beyond the pool_size (default: 5).

Attributes

connection_string

Methods

`<a href="#morpheus.utils.sql_utils.SQLUtils.add_columns_if_not_exists">add_columns_if_not_exists</a>`(table_name, ...[, ...])	Adds columns to a table if they don't already exist.
`<a href="#morpheus.utils.sql_utils.SQLUtils.close">close</a>`()	Closes the database connection.
`<a href="#morpheus.utils.sql_utils.SQLUtils.connect">connect</a>`()	Establishes a connection to the database using the specified SQL configuration.
`<a href="#morpheus.utils.sql_utils.SQLUtils.create_table">create_table</a>`(table_name, columns)	Creates a table in the database.
`<a href="#morpheus.utils.sql_utils.SQLUtils.drop_table">drop_table</a>`(table_name)	Drops a table from the database if it exists.
`<a href="#morpheus.utils.sql_utils.SQLUtils.execute_queries_in_transaction">execute_queries_in_transaction</a>`(queries[, params])	Executes multiple SQL queries within a transaction.
`<a href="#morpheus.utils.sql_utils.SQLUtils.execute_query">execute_query</a>`(query[, params])	Executes a SQL query without returning any result.
`<a href="#morpheus.utils.sql_utils.SQLUtils.execute_query_with_result">execute_query_with_result</a>`(query[, params])	Executes a SQL query and returns the result as a list of rows.
`<a href="#morpheus.utils.sql_utils.SQLUtils.gen_placeholder_str">gen_placeholder_str</a>`(count)	Generates the placeholder string for SQL query parameters.
`<a href="#morpheus.utils.sql_utils.SQLUtils.gen_sql_conn_str">gen_sql_conn_str</a>`(sql_config)	Generates the SQL connection string based on the SQL configuration.
`<a href="#morpheus.utils.sql_utils.SQLUtils.get_existing_column_rows">get_existing_column_rows</a>`(table_name)	Retrieves the existing column names of a table.
`<a href="#morpheus.utils.sql_utils.SQLUtils.get_table_columns">get_table_columns</a>`(table_name)	Retrieves the column names of a table.
`<a href="#morpheus.utils.sql_utils.SQLUtils.insert_data">insert_data</a>`(table_name, values[, columns])	Inserts data into a table.
`<a href="#morpheus.utils.sql_utils.SQLUtils.to_dataframe">to_dataframe</a>`(query[, params])	Executes a SQL query and returns the result as a Pandas DataFrame.
`<a href="#morpheus.utils.sql_utils.SQLUtils.to_table">to_table</a>`(df, table_name[, if_exists, index])	Converts a Pandas DataFrame to a SQL table.

add_columns_if_not_exists(table_name, columns, data_type, default_value=None)[source]

Adds columns to a table if they don’t already exist.

Parameters

table_namestr: The name of the table.
columnstyping.List[str]: The column names to add.
data_typestr: The data type of the columns.
default_valuetyping.Union[int, str], optional: The default value for the columns (default: None).

Returns

new_col_addedbool: True if columns were added, False otherwise.

close()[source]

Closes the database connection.

connect()[source]

Establishes a connection to the database using the specified SQL configuration.

create_table(table_name, columns)[source]

Creates a table in the database.

Parameters

table_namestr: The name of the table to create.
columnsstr: The column definitions for the table.

drop_table(table_name)[source]

Drops a table from the database if it exists.

Parameters

table_namestr: The name of the table to drop.

execute_queries_in_transaction(queries, params=())[source]

Executes multiple SQL queries within a transaction.

Parameters

queriestyping.List[str]: The list of SQL queries to execute.
paramstyping.Tuple, optional: The query parameters (default: ()).

execute_query(query, params=())[source]

Executes a SQL query without returning any result.

Parameters

querystr: The SQL query to execute.
paramstyping.Tuple, optional: The query parameters (default: ()).

execute_query_with_result(query, params=())[source]

Executes a SQL query and returns the result as a list of rows.

Parameters

querystr: The SQL query to execute.
paramstyping.Tuple, optional: The query parameters (default: ()).

Returns

rowstyping.List[Row]: The result of the query as a list of rows.

gen_placeholder_str(count)[source]

Generates the placeholder string for SQL query parameters.

Parameters

countint: The number of placeholders to generate.

Returns

placeholdersstr: The generated placeholder string.

Raises

RuntimeError: If the count is invalid.

classmethod gen_sql_conn_str(sql_config)[source]

Generates the SQL connection string based on the SQL configuration.

Parameters

sql_configtyping.Dict: Configuration parameters for the SQL connection.

Returns

connection_stringstr: The generated SQL connection string.

Raises

RuntimeError: If required SQL parameters are missing or invalid.

get_existing_column_rows(table_name)[source]

Retrieves the existing column names of a table.

Parameters

table_namestr: The name of the table.

Returns

existing_columnstyping.List[str]: The existing column names.

Raises

exc.SQLAlchemyError: If an error occurs while retrieving the existing columns.

get_table_columns(table_name)[source]

Retrieves the column names of a table.

Parameters

table_namestr: The name of the table.

Returns

columnstyping.List[str]: The column names.

Raises

exc.SQLAlchemyError: If an error occurs while retrieving the columns.

insert_data(table_name, values, columns=None)[source]

Inserts data into a table.

Parameters

table_namestr: The name of the table to insert data into.
valuestyping.Tuple: The values to insert.
columnstyping.List[str], optional: The column names to insert data into (default: None).

to_dataframe(query, params=None)[source]

Executes a SQL query and returns the result as a Pandas DataFrame.

Parameters

querystr: The SQL query to execute.
paramstyping.Tuple, optional: The query parameters (default: None).

Returns

dfpd.DataFrame: The result of the query as a Pandas DataFrame.

Raises

exc.SQLAlchemyError: If an error occurs while executing the query or converting the result to a DataFrame.

to_table(df, table_name, if_exists='replace', index=False)[source]

Converts a Pandas DataFrame to a SQL table.

Parameters

dfpd.DataFrame: The DataFrame to convert to a table.
table_namestr: The name of the table.
if_existsstr, optional: Action to take if the table already exists (default: “replace”).
indexbool, optional: Whether to include the DataFrame index as a column in the table (default: False).

Raises

exc.SQLAlchemyError: If an error occurs while converting the DataFrame to a table.

Previous morpheus.utils.sql_utils

Next morpheus.utils.type_aliases