lce_service.database module

lce_service.database.execute_query(dbc: silo_cursor, query: str, params: List[Any] | None, table: str) → int

Execute a database query with error handling and context-aware exception reporting.

This function serves as a wrapper around the database cursors execute method, providing standardized error handling and context-specific error messages. When a database operation fails, it raises a DatabaseExecutionError with table-specific context information to aid in debugging and error tracking.

Parameters:

• dbc – A silo_cursor instance used to execute the database query against the target database. This cursor should be properly configured and connected to the database before calling this function.

• query – The SQL query string to be executed. This can be any valid SQL statement including SELECT, INSERT, UPDATE, DELETE, or DDL operations. The query may contain parameter placeholders (%s) for parameterized execution.

• params – An optional list of parameter values to be bound to the query placeholders. The number and order of parameters must match the placeholders in the query string. Can be an empty list if the query contains no parameters.

• table – The name of the database table being operated on, used for context in error messages. This should include the schema prefix if applicable (e.g., “master.`dynamic_app`”).

Returns:

The number of affected rows from the database operation. For INSERT, UPDATE, and DELETE operations, this represents the count of rows that were modified.

Raises:

DatabaseExecutionError – Raised when the database query execution fails for any reason (syntax errors, constraint violations, connection issues, etc.). The error message includes the table name for context, and the original exception is chained for detailed debugging information.

lce_service.database.get_column_from_table(dbc: silo_cursor, query: str, params: List[Any] | None, table: str) → List[Any]

Execute a database query and retrieve a single column of results with error handling.

This function serves as a wrapper around the database cursor’s autofetch_column method, providing standardized error handling and context-aware exception reporting. When a database operation fails, it raises a DatabaseExecutionError with table-specific context information to aid in debugging and error tracking.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch column data from the target database. This cursor should be properly configured and connected to the database before calling this function.

• query – The SQL query string to be executed that should return results from which a single column will be extracted. The query may contain parameter placeholders (%s) for parameterized execution.

• params – An optional list of parameter values to be bound to the query placeholders. The number and order of parameters must match the placeholders in the query string. Can be an empty list if the query contains no parameters.

• table – The name of the database table being queried, used for context in error messages. This should include the schema prefix if applicable (e.g., “master.`dynamic_app`”).

Returns:

A list containing all values from the first column of the query results. The list may be empty if the query returns no rows. The data types of the returned values depend on the column being queried.

Raises:

DatabaseExecutionError – Raised when the database query execution fails for any reason (syntax errors, connection issues, invalid parameters, etc.). The error message includes the table name for context, and the original exception is chained for detailed debugging information.

lce_service.database.get_fields_from_table(dbc: silo_cursor, query: str, params: List[Any] | None, table: str) → List[Dict[str, Any]]

Execute a database query and retrieve all rows as dictionaries with error handling.

This function serves as a wrapper around the database cursor’s autofetchall_dict method, providing standardized error handling and context-aware exception reporting. When a database operation fails, it raises a DatabaseExecutionError with table-specific context information to aid in debugging and error tracking.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch all rows as dictionaries from the target database. This cursor should be properly configured and connected to the database before calling this function.

• query – The SQL query string to be executed that should return multiple rows of data. The query may contain parameter placeholders (%s) for parameterized execution.

• params – An optional list of parameter values to be bound to the query placeholders. The number and order of parameters must match the placeholders in the query string. Can be an empty list if the query contains no parameters.

• table – The name of the database table being queried, used for context in error messages. This should include the schema prefix if applicable (e.g., “master.`dynamic_app`”).

Returns:

A list of dictionaries where each dictionary represents a row from the query results, with column names as keys and column values as values. Returns an empty list if the query returns no rows.

Raises:

DatabaseExecutionError – Raised when the database query execution fails for any reason (syntax errors, connection issues, invalid parameters, etc.). The error message includes the table name for context, and the original exception is chained for detailed debugging information.

lce_service.database.get_in_clause(column: str, values: List[Any]) → str

Build a parameterized SQL IN condition for a column.

Parameters:

• column – The fully qualified and quoted column (e.g. “dao.`app_id`”).

• values – The values to match. One placeholder is generated per value.

Returns:

The SQL condition with placeholders for each value.

lce_service.database.get_prepared_db_params(params: List[Any] | None, strict: bool = False) → List[Any] | None

Prepare database parameters by extracting values from enums and handling string encoding.

This function processes a list of parameters intended for database operations by extracting the ‘value’ attribute from objects that have one, while leaving other parameters unchanged. Additionally, it ensures string parameters are properly encoded for latin1 compatibility. This is commonly used to handle enum objects or other wrapper objects that store their actual database value in a ‘value’ attribute, ensuring that the correct primitive values are passed to database operations.

Parameters:

• params – An optional list of parameter values that may include objects with ‘value’ attributes (such as enum instances) or primitive values. Can be None if no parameters are needed for the database operation.

• strict – A boolean flag that controls string encoding behavior. When True, raises UnicodeEncodeError for characters that cannot be encoded in latin1. When False (default), replaces problematic characters with replacement characters to ensure compatibility. Defaults to False for more lenient encoding handling.

Returns:

A list of prepared parameter values where objects with ‘value’ attributes have been replaced with their extracted values, and string parameters have been processed for latin1 encoding compatibility. Other parameters remain unchanged. Returns None if the input params is None, maintaining the same None-handling behavior for database operations that don’t require parameters.

Raises:

InvalidCharacterError – Raised when strict=True and a string parameter contains characters that cannot be encoded in latin1. When strict=False, problematic characters are replaced instead of raising an error.

lce_service.database.get_row_from_table(dbc: silo_cursor, query: str, params: List[Any] | None, table: str) → Dict[str, Any]

Execute a database query and retrieve a single row as a dictionary with error handling.

This function serves as a wrapper around the database cursor’s autofetchrow_dict method, providing standardized error handling and context-aware exception reporting. When a database operation fails, it raises a DatabaseExecutionError with table-specific context information to aid in debugging and error tracking.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch a single row as a dictionary from the target database. This cursor should be properly configured and connected to the database before calling this function.

• query – The SQL query string to be executed that should return a single row of data. The query may contain parameter placeholders (%s) for parameterized execution.

• params – An optional list of parameter values to be bound to the query placeholders. The number and order of parameters must match the placeholders in the query string. Can be an empty list if the query contains no parameters.

• table – The name of the database table being queried, used for context in error messages. This should include the schema prefix if applicable (e.g., “master.`dynamic_app`”).

Returns:

A dictionary representing a single row from the query results, with column names as keys and column values as values. Returns None if the query returns no rows.

Raises:

DatabaseExecutionError – Raised when the database query execution fails for any reason (syntax errors, connection issues, invalid parameters, etc.). The error message includes the table name for context, and the original exception is chained for detailed debugging information.

lce_service.database.get_value_from_table(dbc: silo_cursor, query: str, params: List[Any] | None, table: str) → Any

Execute a database query and retrieve a single value result with error handling.

This function serves as a wrapper around the database cursor’s autofetch_value method, providing standardized error handling and context-aware exception reporting. When a database operation fails, it raises a DatabaseExecutionError with table-specific context information to aid in debugging and error tracking.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch a single value from the target database. This cursor should be properly configured and connected to the database before calling this function.

• query – The SQL query string to be executed that should return a single value result. The query may contain parameter placeholders (%s) for parameterized execution.

• params – An optional list of parameter values to be bound to the query placeholders. The number and order of parameters must match the placeholders in the query string. Can be an empty list if the query contains no parameters.

• table – The name of the database table being queried, used for context in error messages. This should include the schema prefix if applicable (e.g., “master.`dynamic_app`”).

Returns:

A single value from the first row and first column of the query results. Returns None if the query returns no rows. The data type of the returned value depends on the column being queried.

Raises:

DatabaseExecutionError – Raised when the database query execution fails for any reason (syntax errors, connection issues, invalid parameters, etc.). The error message includes the table name for context, and the original exception is chained for detailed debugging information.