lce_service.dynamic_app_api.core module

class lce_service.dynamic_app_api.core.BulkSQLOperationComponents(set_clause: str, where_clause: str, params: List[Any])

Bases: NamedTuple

Components for bulk database operations.

Variables:

• set_clause – The SET clause portion of the SQL operation (without the “SET” keyword) containing either simple assignments or CASE statements for updating specific columns.

• where_clause – The WHERE clause portion of the SQL operation (without the “WHERE” keyword) that defines the conditions for selecting records.

• params – Parameter values in the correct order to be bound to the placeholders (%s) in both the SET clauses and WHERE clause. This includes field values, user ID for audit fields, and key values for record identification.

params: List[Any]

Alias for field number 2

set_clause: str

Alias for field number 0

where_clause: str

Alias for field number 1

exception lce_service.dynamic_app_api.core.DatabaseExecutionError(message: str)

Bases: Exception

Exception raised when a database operation fails to execute successfully.

This error is raised when attempting to execute a database operation (such as INSERT, UPDATE, DELETE, or SELECT queries) that encounters execution failures due to database connectivity issues, constraint violations, syntax errors, or other database-level problems.

Parameters:

message – A descriptive error message explaining what database operation failed and any relevant details about the execution error.

class lce_service.dynamic_app_api.core.DynamicAppEntityRelationship(collection_object_id: int, entity_id: int)

Bases: NamedTuple

Information about Dynamic Application key relationships for database operations.

This NamedTuple represents the key relationship data that links collection objects to other entities (such as presentation objects or alert objects) within a Dynamic Application. It stores the identifiers needed to establish and maintain these relationships in the master.dynamic_app_keys table.

Variables:

• collection_object_id – The unique integer identifier of the collection object that is being referenced or linked to another entity. This corresponds to the obj_id in the database.

• entity_id – The unique integer identifier of the entity (presentation object, alert object, etc.) that references the collection object. This corresponds to the entity_id in the database.

collection_object_id: int

Alias for field number 0

entity_id: int

Alias for field number 1

class lce_service.dynamic_app_api.core.ExistingAndNewRecords(existing_records: Dict[Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)], EnumLabelMixin], new_records: Dict[Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)], EnumLabelMixin], guid_mapping: Dict[Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)], Annotated[str | None, BeforeValidator(func=validate_optional_string, json_schema_input_type=PydanticUndefined)]])

Bases: NamedTuple

Information about records categorized as existing or new for database operations.

Variables:

• existing_records – Dictionary mapping existing record GUIDs to their corresponding model instances that were found in the database.

• new_records – Dictionary mapping newly generated GUIDs to model instances for records that don’t exist in the database.

• guid_mapping – Dictionary mapping newly generated GUIDs to their provided identifiers (original GUID or reference value), enabling tracking of GUID changes during database operations.

existing_records: Dict[Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)], EnumLabelMixin]

Alias for field number 0

guid_mapping: Dict[Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)], Annotated[str | None, BeforeValidator(func=validate_optional_string, json_schema_input_type=PydanticUndefined)]]

Alias for field number 2

new_records: Dict[Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)], EnumLabelMixin]

Alias for field number 1

exception lce_service.dynamic_app_api.core.InvalidCharacterError(message: str)

Bases: InvalidResourceError, DatabaseExecutionError

Exception raised when a value contains characters that cannot be stored in the database.

The database stores text as latin1, so values containing characters outside of latin1 are rejected. This is a user input error, so it is an InvalidResourceError, while remaining a DatabaseExecutionError for callers that handle failed database operations.

Parameters:

message – A descriptive error message identifying the invalid character and value.

exception lce_service.dynamic_app_api.core.InvalidResourceError(message: str)

Bases: AttributeError

Exception raised when a resource is found but contains invalid or malformed data.

This error is raised when attempting to validate a resource (such as a Dynamic Application) that exists in the database but fails validation due to invalid data, incorrect format, or constraint violations.

Parameters:

message – A descriptive error message explaining what validation failed and any relevant details about the invalid data.

exception lce_service.dynamic_app_api.core.MissingResourceError(message: str, matches: List[str] = None)

Bases: LookupError

Exception raised when a requested resource cannot be found.

This error is raised when attempting to retrieve a resource (such as a Dynamic Application) that does not exist in the database or is otherwise unavailable.

Parameters:

• message – A descriptive error message explaining which resource is missing and any relevant identifying information.

• matches – An optional list of entity names that matched the search criteria, allowing the caller to see all potential matches and potentially refine their search to be more specific.

lce_service.dynamic_app_api.core.create_reference_map_to_entities(entities: List[EntityIdentifier]) → Dict[str, EntityIdentifier]

Create a mapping dictionary from entity references to EntityIdentifier instances.

This function builds a lookup dictionary that maps entity references to their corresponding EntityIdentifier instances. For entities that have a reference, the mapping uses the reference as the key. For entities without a reference, the mapping uses the current GUID as the key. This allows for efficient lookup of entities by either their reference or their current GUID.

Parameters:

entities – A list of EntityIdentifier instances to be mapped. Each EntityIdentifier should have a guid attribute and may optionally have a reference attribute indicating the entity’s identifier.

Returns:

A dictionary mapping entity references to EntityIdentifier instances. Keys are either reference values or current GUIDs. Values are the corresponding EntityIdentifier instances. This mapping enables efficient lookup of entities by their reference identifiers.

lce_service.dynamic_app_api.core.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.dynamic_app_api.core.generate_guid() → Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)]

Generate a unique GUID using UUID4 and truncate to maintain 32-character format.

Uses Python’s uuid.uuid4() which provides 122 bits of randomness from the OS’s cryptographically secure random number generator, then converts to hexadecimal and truncates to 32 characters for compatibility. This provides FIPS compliance while maintaining the expected format and superior collision resistance.

Returns:

A 32-character uppercase hexadecimal string representing the generated GUID.

lce_service.dynamic_app_api.core.get_app_id_from_dynamic_app_guid(dbc: silo_cursor, app_guid: str) → int

Retrieves the application ID for a Dynamic Application using its GUID.

This function queries the database to find the numeric application ID (aid) that corresponds to a given Dynamic Application GUID. It uses the database manager context to execute the query and handles cases where no matching application is found.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch application data from the master database.

• app_guid – The GUID (Globally Unique Identifier) of the Dynamic Application to look up in the database.

Returns:

The numeric application ID (aid) associated with the provided GUID

Raises:

• MissingResourceError – If no Dynamic Application is found with the specified GUID in the database

• DatabaseExecutionError – Raised when the database query execution fails for any reason

lce_service.dynamic_app_api.core.get_app_id_from_dynamic_app_name(dbc: silo_cursor, name: str | None) → int | None

Retrieves the application ID for a Dynamic Application using its name with fuzzy matching.

This function queries the database to find the numeric application ID (aid) that corresponds to a given Dynamic Application name. It uses fuzzy string matching to handle minor variations in application names and provides intelligent disambiguation when multiple similar matches are found. The function delegates to get_id_from_entity_name for the actual fuzzy matching logic and database operations.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch application data from the master database. Must be properly configured and connected to the database before calling this function.

• name – The name of the Dynamic Application to look up. Can be None, which will raise a ValueError. Must be a non-empty string between 1 and 64 characters if provided. Leading and trailing whitespace will be stripped during processing.

Returns:

The numeric application ID (aid) associated with the best matching Dynamic Application name if a confident match is found. Returns None if no confident match can be determined (though in practice, exceptions are raised for all failure cases).

Raises:

• ValueError – When the name parameter is None, empty, not a string, longer than 64 characters, or when multiple Dynamic Applications have similar scores making the match ambiguous (score difference less than FUZZY_MATCHING_CONFIDENT_SCORE_DIFFERENCE points).

• MissingResourceError – When no Dynamic Applications in the database match the provided name above the minimum score threshold (FUZZY_MATCHING_MIN_SCORE), or when the closest match falls below the acceptable similarity threshold. Also when multiple Dynamic Applications have similar scores making the match ambiguous (score difference less than FUZZY_MATCHING_CONFIDENT_SCORE_DIFFERENCE points).

• DatabaseExecutionError – Raised when the database query execution fails for any reason (connection issues, invalid table/column names, syntax errors, etc.).

lce_service.dynamic_app_api.core.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.dynamic_app_api.core.get_components_for_bulk_update(model_info: EntityModelInformation, records_by_key: Dict[Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)], EnumLabelMixin], table: str, user: User) → BulkSQLOperationComponents

Generate SQL SET clauses and parameters for bulk update operations with audit fields.

This function constructs the SET portion of an SQL UPDATE statement for bulk update operations, handling both single-record updates (using simple assignments) and multi-record updates (using CASE statements). It respects immutable column constraints, automatically adds appropriate audit fields based on the target table schema, and optimizes the SQL generation based on whether one or multiple records are being updated.

The function generates different SQL patterns: - Single record: column = %s - Multiple records: column = CASE key_column WHEN %s THEN %s … ELSE column END

Parameters:

• model_info – An EntityModelInformation instance containing metadata about the entity structure, including column_mapping (API field to database column mapping), immutable_fields (fields that cannot be updated), and primary_db_key (the database column name used for identification in CASE statements).

• records_by_key – A dictionary mapping entity GUIDs to their corresponding model instances that contain the field values to be updated. Each model instance should have attributes corresponding to the API field names in column_mapping.

• table – The name of the database table being updated, used to determine the appropriate audit field schema. Different tables use different audit field naming conventions (e.g., DYNAMIC_APP_TABLE uses date_edit/user_edit, while others use edit_date/edit_user).

• user – The User instance representing the user performing the update operation, used to populate audit fields in the SET clauses for tracking who made the changes.

Returns:

A BulkSQLOperationComponents instance containing: - set_clause: Comma-separated string of SQL SET assignments ready for use in

an UPDATE statement

• where_clause: SQL WHERE clause for identifying the records to update

• params: List of parameter values in the correct order to be bound to the

  placeholders (%s) in both SET and WHERE clauses

lce_service.dynamic_app_api.core.get_dynamic_application_header(dynamic_app: DynamicApp) → str

Generate a standardized header string for Dynamic Application identification.

This function creates a formatted string that provides comprehensive identification information for a Dynamic Application, including its name, database ID, and GUID. The header format is standardized across the codebase for consistent error messages, logging, and user-facing output when referencing Dynamic Applications.

Parameters:

dynamic_app – A DynamicApp instance representing the Dynamic Application for which to generate the identification header. The function uses the name, id, and guid attributes to construct the formatted identification string.

Returns:

A formatted string containing the Dynamic Application’s name, database ID, and GUID in the standardized format: “Dynamic Application ‘name’ (id) [GUID=guid]”. This string provides complete identification information for the application and can be used in error messages, logs, and other user-facing output.

lce_service.dynamic_app_api.core.get_entities_from_db(dbc: silo_cursor, params: List[Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)]], where_clause: str, entity_columns: Dict[str, str], table: str, is_new: bool, guid_mapping: Dict[Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)], Annotated[str | None, BeforeValidator(func=validate_optional_string, json_schema_input_type=PydanticUndefined)]]) → List[EntityIdentifier]

Retrieve and validate entity records from the database as EntityIdentifier instances.

This function executes a database query to fetch entity records from the specified table using the provided WHERE clause and parameters. The retrieved data is then validated and converted into EntityIdentifier model instances. This function serves as a standardized way to retrieve entity information with proper validation and error handling.

Parameters:

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

• params – A list of GUID parameters to be bound to the query placeholders in the WHERE clause. The number and order of parameters must match the placeholders in the where_clause string.

• where_clause – The WHERE clause portion of the SQL query (without the “WHERE” keyword) that defines the conditions for selecting records. Should contain parameter placeholders (%s) that correspond to the params list.

• entity_columns – A dictionary mapping API field names to database column names for the fields that should be retrieved and included in the EntityIdentifier instances. Used to construct the SELECT portion of the query with appropriate column aliasing.

• table – The name of the database table to query, including schema prefix if applicable (e.g., “master.`dynamic_app`”). Used for both the query construction and error context in exception messages.

• is_new – A boolean flag indicating whether the retrieved entities should be marked as newly created. When True, the ‘new’ attribute of each EntityIdentifier instance is set to True to indicate the entity was just created. When False, entities are marked as existing records.

• guid_mapping – A dictionary mapping entity GUIDs to their original identifiers (old GUIDs or reference values) for tracking GUID changes during database operations. Used to populate the reference field in returned EntityIdentifier instances for audit and tracking purposes.

Returns:

A list of validated EntityIdentifier model instances representing the retrieved entity records. Each instance contains the entity’s identifier information (id, guid, app_id, etc.) as defined by the entity_columns mapping. Returns an empty list if no records match the query conditions.

Raises:

• InvalidResourceError – Raised when any retrieved entity data fails validation against the EntityIdentifier model schema, indicating invalid or malformed data in the database.

• DatabaseExecutionError – Raised when the database query execution fails for any reason (syntax errors, connection issues, invalid parameters, etc.), with the original exception chained for debugging purposes.

lce_service.dynamic_app_api.core.get_event_id_from_event_policy_guid(dbc: silo_cursor, event_guid: str) → int

Retrieves the event policy ID for an Event Policy using its GUID.

This function queries the database to find the numeric event policy ID that corresponds to a given Event Policy GUID. It uses the database manager context to execute the query and handles cases where no matching event policy is found.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch event policy data from the master database.

• event_guid – The GUID (Globally Unique Identifier) of the Event Policy to look up in the database.

Returns:

The numeric event policy ID associated with the provided GUID

Raises:

• MissingResourceError – If no Event Policy is found with the specified GUID in the database

• DatabaseExecutionError – Raised when the database query execution fails for any reason

lce_service.dynamic_app_api.core.get_event_id_from_event_policy_name(dbc: silo_cursor, name: str | None) → int | None

Retrieves the event policy ID for an Event Policy using its name with fuzzy matching.

This function queries the database to find the numeric event policy ID that corresponds to a given Event Policy name. It uses fuzzy string matching to handle minor variations in event policy names and provides intelligent disambiguation when multiple similar matches are found. The function delegates to get_id_from_entity_name for the actual fuzzy matching logic and database operations.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch event policy data from the master database. Must be properly configured and connected to the database before calling this function.

• name – The name of the Event Policy to look up. Can be None, which will raise a ValueError. Must be a non-empty string between 1 and 64 characters if provided. Leading and trailing whitespace will be stripped during processing.

Returns:

The numeric event policy ID associated with the best matching Event Policy name if a confident match is found. Returns None if no confident match can be determined (though in practice, exceptions are raised for all failure cases).

Raises:

• ValueError – When the name parameter is None, empty, not a string, longer than 64 characters, or when multiple Event Policies have similar scores making the match ambiguous (score difference less than FUZZY_MATCHING_CONFIDENT_SCORE_DIFFERENCE points).

• MissingResourceError – When no Event Policies in the database match the provided name above the minimum score threshold (FUZZY_MATCHING_MIN_SCORE), or when the closest match falls below the acceptable similarity threshold. Also when multiple Event Policies have similar scores making the match ambiguous (score difference less than FUZZY_MATCHING_CONFIDENT_SCORE_DIFFERENCE points).

• DatabaseExecutionError – Raised when the database query execution fails for any reason (connection issues, invalid table/column names, syntax errors, etc.).

lce_service.dynamic_app_api.core.get_existing_and_new_records(dbc: silo_cursor, records: List[EnumLabelMixin], table: str, model_info: EntityModelInformation) → ExistingAndNewRecords

Identify existing and new records using GUID identifiers.

This function processes a list of records to determine which ones already exist in the database and which ones are new. For new records, it generates new GUIDs and maintains a mapping from the original identifier (if provided) to the newly generated GUID. This separation enables efficient bulk database operations by allowing existing records to be updated and new records to be inserted using optimized SQL statements.

The function performs the following operations: 1. Groups records by whether they have GUIDs or not for batch processing 2. Executes batch queries to find existing records using GUID lookups 3. Generates new GUIDs for records that don’t exist in the database 4. Maintains mapping between original identifiers and newly generated GUIDs for tracking

Parameters:

• dbc – A silo_cursor instance used to execute database queries for record lookup from the target database. This cursor should be properly configured and connected to the database before calling this function.

• records – A list of EnumLabelMixin model instances to be processed and categorized as existing or new. Each record may have a guid attribute for identification and a reference attribute for tracking purposes.

• table – The name of the database table to query for existing records, including schema prefix if applicable (e.g., “master.`dynamic_app`”). Used in the SELECT query to check for existing GUIDs.

• model_info – An EntityModelInformation instance containing metadata about the entity structure, including the primary_db_key field that specifies the database column name used for GUID lookups (e.g., “app_guid”, “req_guid”). This column is used in the WHERE clause of the lookup query.

Returns:

An ExistingAndNewRecords instance containing three dictionaries: - existing_records: Dict[GUID, EnumLabelMixin] mapping existing record GUIDs

to their corresponding model instances that were found in the database

• new_records: Dict[GUID, EnumLabelMixin] mapping newly generated GUIDs to

  model instances for records that don’t exist in the database or don’t have GUIDs

• guid_mapping: Dict[GUID, OptionalString] mapping newly generated GUIDs to

  their original identifiers, enabling tracking of GUID changes during database operations

Raises:

• DatabaseExecutionError – Raised when any database query execution fails during the record lookup process, with the original exception chained for debugging purposes.

• InvalidResourceError – Raised when any entity is attempted to be updated that belongs to a different Dynamic Application than the one being updated. Also raised when any new entity has a GUID or ID set.

lce_service.dynamic_app_api.core.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.dynamic_app_api.core.get_identifier_for_entity(dbc: silo_cursor, entity_identifier: int | str, id_from_guid_retriever: Callable, id_from_name_retriever: Callable) → int

Resolve an entity identifier to its integer ID, handling ID, GUID, and name inputs.

This function normalizes entity identification by accepting either a direct integer ID, a GUID string, or an entity name string. When an integer is provided, it’s returned directly. When a GUID string is provided, it’s validated and then resolved to its corresponding integer ID using the provided GUID retriever function. When a name string is provided (after GUID validation fails), it’s resolved using the provided name retriever function with fuzzy matching capabilities.

Parameters:

• dbc – A silo_cursor instance used to execute database queries when resolving a GUID or name to its integer ID.

• entity_identifier – The identifier for the entity, which can be either: - An integer representing the direct entity ID - A string representing the entity’s GUID - A string representing the entity’s name (for fuzzy matching)

• id_from_guid_retriever – A callable function that takes a silo_cursor and a GUID string as parameters and returns the corresponding integer ID for the entity. This function is only called when entity_identifier is a valid GUID string. The retriever function must accept 2 arguments: a dbc instance and a GUID string. It must return the integer ID for the corresponding entity or raise a MissingResourceError if it cannot be found.

• id_from_name_retriever – A callable function that takes a silo_cursor and a name string as parameters and returns the corresponding integer ID for the entity using fuzzy matching. This function is only called when entity_identifier fails GUID validation. The retriever function must accept 2 arguments: a dbc instance and a name string. It must return the integer ID for the best matching entity or raise appropriate exceptions if no match is found.

Returns:

The integer ID of the entity corresponding to the provided identifier.

Raises:

• MissingResourceError – When either retriever function is called and determines that no entity exists with the specified identifier in the database.

• ValueError – When an entity identifier is provided that is not a valid GUID (raised by validate_guid during GUID validation) or when name-based matching fails due to ambiguous or invalid names.

• DatabaseExecutionError – Raised when the database query execution fails for any reason during the retrieval process.

lce_service.dynamic_app_api.core.get_library_version_in_collection_env_from_db(dbc: silo_cursor, collection_env: str, underscore_name: str, hyphen_name: str) → Dict[str, int | str]

Retrieve library version information from the database for a specific execution environment.

This function queries the database to find a library that is aligned to a specific execution environment (collection environment). It searches for the library using both underscore and hyphen naming conventions to handle different package naming formats. The function joins the content_libraries and content_libraries_ppack tables to find libraries that are associated with the specified execution environment.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch library data from the master database.

• collection_env – The GUID of the execution environment to search for aligned libraries.

• underscore_name – The library name using underscore naming convention (e.g., ‘my_library’). This represents one possible naming format for the library.

• hyphen_name – The library name using hyphen naming convention (e.g., ‘my-library’). This represents an alternative naming format for the same library.

Returns:

A dictionary containing the library information if found, with keys: - ‘name’: The library name as stored in the database - ‘version’: The version string of the library - ‘lib_id’: The unique library identifier Returns an empty dictionary if no matching library is found.

Raises:

DatabaseExecutionError – Raised when any database operation fails for any reason

lce_service.dynamic_app_api.core.get_models_from_entities(model: BaseModel, entities: List[Dict[str, Any]]) → List[BaseModel] | None

Convert a list of entity dictionaries into validated Pydantic model instances.

Takes a list of dictionaries representing database entities and transforms them into instances of the specified Pydantic model type. Each dictionary is validated against the model schema during instantiation. Optionally filters out specified keys from all dictionaries before validation. If any validation fails, the function delegates error handling to handle_validation_error, which raises an InvalidResourceError.

Parameters:

• model – The Pydantic BaseModel class to instantiate for each entity dictionary. This model defines the schema and validation rules for the entities.

• entities – A list of dictionaries, where each dictionary contains the data for a single entity to be converted into a model instance.

Returns:

A list of validated Pydantic model instances, one for each entity dictionary in the input list. Returns None if validation fails (though in practice, an InvalidResourceError is raised before None can be returned).

Raises:

InvalidResourceError – Raised when any entity dictionary fails validation against the specified model schema, containing formatted error messages describing all validation failures.

lce_service.dynamic_app_api.core.get_models_from_stored_data(model: Type[EnumLabelMixin], entities: List[Dict[str, Any]]) → List[EnumLabelMixin] | None

Convert stored entity data into model instances, keeping the values that fail validation.

Unlike get_models_from_entities, a stored value that fails the validation of its field does not fail the conversion: it is kept as is and reported in the model’s validation_errors (see EnumLabelMixin.model_validate_stored), so stored entities can always be read.

Parameters:

• model – The model class to instantiate for each entity.

• entities – The stored data of the entities, keyed by field name.

Returns:

A model instance for each entity. Returns None if validation fails (though in practice, an InvalidResourceError is raised before None can be returned).

Raises:

InvalidResourceError – Raised when an entity fails a rule of its model that is not about a single field, such as the supported application types.

lce_service.dynamic_app_api.core.get_pip_config_options(dbc: silo_cursor) → Dict[str, List[str]]

Retrieve and validate pip configuration options from the system settings table.

This function queries the database to fetch the wheel downloader pip options configuration setting from the master.system_settings_core table. These options are used to configure pip behavior for package installation and wheel downloading operations within the system. The function validates that the retrieved options are in valid JSON format and represent a dictionary object before returning them.

Parameters:

dbc – A silo_cursor instance used to execute the database query and fetch the pip configuration options from the master database. Must be properly configured and connected to the database before calling this function.

Returns:

A dictionary containing the pip configuration options as parsed from the wheel_downloader_pip_options column. The structure and content depend on the stored configuration but must be a valid JSON object.

Raises:

• DatabaseExecutionError – Raised when the database query execution fails for any reason (connection issues, invalid table or column names, etc.).

• ValueError – Raised when the retrieved pip options are not in valid JSON format or when the parsed JSON is not a dictionary object. This ensures that only properly formatted configuration data is returned.

lce_service.dynamic_app_api.core.get_powerpack_id_from_guid(dbc: silo_cursor, powerpack_guid: str) → int

Retrieves the numeric ID for a PowerPack using its GUID.

This function queries the database to find the numeric PowerPack ID (ppid) that corresponds to a given PowerPack GUID. It uses the database manager context to execute the query and handles cases where no matching PowerPack is found.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch application data from the master database.

• powerpack_guid – The GUID (Globally Unique Identifier) of the PowerPack to look up in the database.

Returns:

The numeric PowerPack ID (ppid) associated with the provided GUID

Raises:

• MissingResourceError – If no PowerPack is found with the specified GUID in the database

• DatabaseExecutionError – Raised when the database query execution fails for any reason

lce_service.dynamic_app_api.core.get_powerpack_id_from_name(dbc: silo_cursor, name: str | None) → int | None

Retrieves the numeric ID for a PowerPack using its name with fuzzy matching.

This function queries the database to find the numeric PowerPack ID (ppid) that corresponds to a given PowerPack name. It uses fuzzy string matching to handle minor variations in PowerPack names and provides intelligent disambiguation when multiple similar matches are found. The function delegates to get_id_from_entity_name for the actual fuzzy matching logic and database operations.

Parameters:

• dbc – A silo_cursor instance used to execute the database query and fetch PowerPack data from the master database. Must be properly configured and connected to the database before calling this function.

• name – The name of the PowerPack to look up. Must be a non-empty string between 1 and 64 characters. Leading and trailing whitespace will be stripped during processing.

Returns:

The numeric PowerPack ID (ppid) associated with the best matching PowerPack name if a confident match is found.

Raises:

• ValueError – When the name parameter is None, empty, not a string, longer than 64 characters, or when multiple PowerPacks have similar scores making the match ambiguous (score difference less than FUZZY_MATCHING_CONFIDENT_SCORE_DIFFERENCE points).

• MissingResourceError – When no PowerPacks in the database match the provided name above the minimum score threshold (FUZZY_MATCHING_MIN_SCORE). Also when multiple PowerPacks have similar scores making the match ambiguous (score difference less than FUZZY_MATCHING_CONFIDENT_SCORE_DIFFERENCE points).

• DatabaseExecutionError – Raised when the database query execution fails for any reason (connection issues, invalid table/column names, syntax errors, etc.).

lce_service.dynamic_app_api.core.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.dynamic_app_api.core.get_sql_select_statements(mapping: Dict[str, str], alias: str | None = None) → str

Generate SQL SELECT clause statements from a field mapping dictionary.

Constructs a comma-separated string of SQL SELECT statements based on the provided mapping between API field names and database column names. When API field names differ from database column names, appropriate SQL aliases are generated. Optionally supports table aliasing for use in JOIN queries.

Parameters:

• mapping – A dictionary mapping API field names (keys) to database column names (values). Used to generate SELECT statements with appropriate aliasing.

• alias – Optional table alias to prefix column names with (e.g., ‘dao’ for ‘dao.column’). If None, column names are used without a table prefix. Defaults to None.

Returns:

A comma-separated string of SQL SELECT statements, properly formatted with backtick-quoted identifiers and AS aliases where API field names differ from database column names.

lce_service.dynamic_app_api.core.get_total_entity_count(dynamic_app_update_result: DynamicAppUpdateResult) → int

Get the total number of entities represented in a DynamicAppUpdateResult.

Counts the Dynamic Application itself plus the number of child entities across all tracked entity types (snippets, collection objects, presentation objects, thresholds).

Parameters:

dynamic_app_update_result – The update result container holding identifiers for the dynamic app and its child entities.

Returns:

Total count of entities across all types in the update result.

lce_service.dynamic_app_api.core.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.

lce_service.dynamic_app_api.core.handle_validation_error(validation_err: ValidationError) → None

Process Pydantic validation errors and raise an InvalidResourceError.

Extracts all validation errors from a Pydantic ValidationError, formats them into human-readable messages with field paths, and raises a single InvalidResourceError containing all error messages.

Parameters:

validation_err – A Pydantic ValidationError containing one or more validation failures from model validation.

Raises:

InvalidResourceError – Always raised with a formatted string containing all validation error messages, separated by newlines. Each error message includes the field path and error description.

lce_service.dynamic_app_api.core.increment_powerpack_revision_number(dbc: silo_cursor, user: User, pp_guid: Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)] | None, increment_number: int) → None

Increment the revision number for a PowerPack and update its edit metadata.

This function updates a PowerPack’s revision number by incrementing it by 1 and sets the edit timestamp to the current time and the edit user to the specified user. The function first looks up the PowerPack ID using the provided GUID, then performs the update operation. If the PowerPack GUID is None or the PowerPack is not found, the function returns without performing any operations.

Parameters:

• dbc – A silo_cursor instance used to execute database queries for looking up the PowerPack ID and updating the PowerPack record in the master.powerpack table.

• user – The User instance representing the user performing the update operation, used to populate the user_edit field in the database record for tracking who made the revision increment.

• pp_guid – The GUID (Globally Unique Identifier) of the PowerPack whose revision number should be incremented. If None, the function returns immediately without performing any database operations.

• increment_number – The integer amount to add to the PowerPack’s revision number (e.g., the number of entities updated in a bulk operation).

Raises:

• DatabaseExecutionError – Raised when either the PowerPack lookup query or the revision update query fails for any reason during the database operation process.

• InvalidResourceError – Raised when provided PowerPack GUID cannot be found, indicating there is an issue.

lce_service.dynamic_app_api.core.insert_audit_log(dbc: silo_cursor, message: str, user: User, source: str = 'Snippet Framework API', organization_id: int | None = None) → None

Writes an audit log entry for user actions.

This function creates a standardized audit log entry that includes the username and a descriptive message about the action performed. The log entry is written to the master_biz.organizations_log database table with a fixed source identifier and acknowledgment flag.

Parameters:

• dbc – A silo_cursor instance used to execute the database insert query for writing the audit log entry to the master_biz.organizations_log table.

• message – A descriptive message about the action being audited. This message will be prefixed with the user’s username to create the full audit log entry.

• user – The User instance representing the user performing the action being audited. The username and organization_id from this user object are used to populate the audit log entry.

• source – The source the entry is filed under. Defaults to this service’s own source; an action the platform audits under a source of its own, such as running a device tool, is filed under that source instead so both read alike.

• organization_id – The organization the entry is filed under, when it belongs to the organization of what was acted on rather than to the user’s own.

Raises:

DatabaseExecutionError – Raised when the database insert operation fails for any reason during the audit log creation process.

lce_service.dynamic_app_api.core.remove_library_from_execution_environment(dbc: silo_cursor, collection_env: str, content_library_id: int) → None

Removes the alignment between a library and an execution environment.

This function removes the association between a specific library and an execution environment by deleting the corresponding record from the content_libraries_ppack table. This effectively un-aligns the library from the execution environment, making it unavailable for use in that environment.

Parameters:

• dbc – The database cursor object used to execute the SQL query

• collection_env – The GUID of the execution environment to un-align the library from

• content_library_id – The unique identifier of the library to un-align

Raises:

DatabaseExecutionError – Raised when the database delete operation fails for any reason

lce_service.dynamic_app_api.core.replace_reference_identifiers(entities: List[EnumLabelMixin], reference_map: Dict[str, str]) → None

Replace entity reference identifiers in specified fields with formatted entity references.

This function processes a list of entities and replaces entity reference identifiers found in a specified field (typically formulas) with properly formatted entity references. It uses regular expressions to find and replace identifiers, ensuring that longer identifiers are matched before shorter ones to prevent partial matches.

The function is commonly used to update formula references in presentation objects when collection object or threshold object identifiers, ensuring that formulas continue to reference the correct entities after GUID remapping or entity creation.

Parameters:

• entities – A list of EnumLabelMixin model instances that contain fields with entity reference identifiers to be replaced. Each entity should have the field specified by field_name that may contain reference identifiers.

• reference_map – A dictionary mapping original entity references (either old GUIDs or reference strings) to their corresponding identifier. This mapping enables lookup of the current entity information for replacement purposes.

lce_service.dynamic_app_api.core.reset_collection_object_usage_type(dbc: silo_cursor, obj_id: int, app_id: int, user: User) → None

Reset a collection object’s usage type to STANDARD and update audit fields.

This function updates a specific collection object within a Dynamic Application to reset its usage type to the STANDARD value. It also updates the audit fields (edit_user and edit_date) to track when and by whom the change was made. This operation is typically used to normalize collection objects that may have had special usage types assigned.

Parameters:

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

• obj_id – The unique integer identifier of the collection object whose usage type should be reset. This corresponds to the primary key of the collection object in the dynamic_app_objects table.

• app_id – The unique integer identifier of the Dynamic Application that owns the collection object. This is used as an additional constraint to ensure the operation only affects collection objects within the correct application context and prevents cross-application modifications.

• user – The User instance representing the user performing the reset operation, used to populate the edit_user audit field for tracking who made the usage type change.

Raises:

DatabaseExecutionError – Raised when the database UPDATE operation fails for any reason during the usage type reset process, as propagated from the underlying execute_query function.

lce_service.dynamic_app_api.core.set_entity_fields(app_id: int, app_guid: Annotated[str, BeforeValidator(func=validate_guid, json_schema_input_type=PydanticUndefined)], entities: List[DynamicAppEntity]) → None

Prepare entity objects for database by setting required fields and generating GUIDs.

This function prepares a list of Dynamic Application entities for database insertion by setting the necessary fields that link them to their parent Dynamic Application and conditionally generating unique GUIDs for each entity when creating a new application. It modifies the entities in-place to ensure they have all required fields populated before database operations.

The function performs the following operations for each entity: 1. Conditionally generates a new unique GUID using the generate_guid() function if this

is for a new application

2. Sets the app_id to link the entity to its parent Dynamic Application

3. Sets the app_guid to provide the GUID reference to the parent Dynamic Application

Parameters:

• app_id – The unique integer identifier (aid) of the Dynamic Application that owns these entities. This value is assigned to each entity’s app_id field to establish the parent-child relationship in the database.

• app_guid – The GUID (app_guid) of the Dynamic Application that owns these entities. This value is assigned to each entity’s app_guid field to provide GUID-based reference to the parent application.

• entities – A list of DynamicAppEntity instances that need to be prepared for database creation. Each entity in this list will be modified in-place to have its guid (conditionally), app_id, and app_guid fields set appropriately.

lce_service.dynamic_app_api.core.update_dynamic_app_keys_table(dbc: silo_cursor, app_id: int, key_type_id: Literal[1, 2], entities: Set[DynamicAppEntityRelationship]) → None

Update dynamic app keys table with relationships between collection objects and entities.

This function establishes relationships in the dynamic app keys table by creating records that link collection objects to newly created entities (such as threshold objects or presentation objects). It performs a bulk INSERT operation to efficiently create multiple relationship records in a single database transaction. The function only operates when there are created entities to process, making it safe to call with empty entity lists.

The function builds a batch INSERT statement with multiple VALUES clauses to optimize database performance when creating multiple relationship records. Each relationship record associates a collection object with a created entity through the specified key_type identifier, enabling proper entity relationships within Dynamic Applications.

Parameters:

• dbc – A silo_cursor instance used to execute the database INSERT query for creating relationship records in the master.dynamic_app_keys table. This cursor should be properly configured and connected to the database before calling this function.

• app_id – The unique integer identifier of the Dynamic Application that owns the collection objects and entities being linked. This value is stored in the app_id column of each relationship record to scope relationships to the correct application context.

• key_type_id – The integer identifier that specifies the type of entity being linked to collection objects (stored in the Z_type column). This value categorizes the relationship type (e.g., threshold objects, presentation objects) and enables proper entity type identification in the keys table.

• entities – A set of DynamicAppEntityRelationship instances containing the relationship data to be inserted into the dynamic app keys table. Each relationship specifies a collection object ID, entity ID, entity type ID, and application ID that define the relationship between collection objects and their associated entities. If the set is empty, no database operations are performed.

Raises:

DatabaseExecutionError – Raised when the database INSERT operation fails for any reason during the relationship creation process, as propagated from the underlying execute_query function.

lce_service.dynamic_app_api.core.validate_model(model_type: BaseModel, model: Dict[str, Any]) → None

Validate a dictionary against a Pydantic model schema.

Attempts to validate the provided dictionary data against the specified Pydantic model type. If validation fails, the function delegates error handling to handle_validation_error, which raises an InvalidResourceError with detailed validation failure information.

Parameters:

• model_type – The Pydantic BaseModel class to validate against.

• model – A dictionary containing the data to be validated against the model schema.

Raises:

InvalidResourceError – Raised when validation fails, containing formatted error messages describing all validation failures.