createIndexCount

createIndexCount is a custom policy check that ensures a table has less than a specified number of indexes.

Learn how to create and customize the createIndexCount Liquibase Custom Policy Check using a Python script.

This example works for relational databases. You can use this check as it is or customize it further to fit your needs in your SQL database.

For a conceptual overview of this feature, see Liquibase Pro Custom Policy Checks.

Scope	Database
database	Relational

Step-by-step

These steps describe how to create the Custom Policy Check. It does not exist by default in Liquibase Pro.

1. Create a Check Settings file: Use the Checks Settings Configuration File

2. Add this code to your Checks Settings file:

   Copy

   createIndexCount Python Script

   ###
   ### This script ensures a table has less than x indexes
   ###
   ### Notes:
   ### 1. Uses liquibase_utilities cache to aggregate index totals across changesets
   ###
   
   ###
   ### Helpers come from Liquibase
   ###
   import liquibase_utilities
   import sys
   
   ###
   ### Functions
   ###
   def find_snapshot_object(object_list, type, key, value):
       """Returns a snapshot object given a key (e.g., name) and attribute."""
       for object in object_list:
           if object[type][key].lower() == value.lower():
               return object
       return None
   
   ###
   ### main
   ###
   
   ###
   ### Retrieve log handler
   ### Ex. liquibase_logger.info(message)
   ###
   liquibase_logger = liquibase_utilities.get_logger()
   
   ###
   ### Retrieve status handler
   ###
   liquibase_status = liquibase_utilities.get_status()
   
   ###
   ### Retrieve JSON snapshot
   ###
   liquibase_snapshot = liquibase_utilities.get_snapshot()
   
   ###
   ### Exit if table data is missing
   ###
   if "liquibase.structure.core.Table" not in liquibase_snapshot["snapshot"]["objects"]:
       liquibase_status.fired = False
       liquibase_logger.warning("Table data missing from snapshot. Check skipped.")
       sys.exit(1)
   
   ###
   ### Retrieve columns and tables from snapshot
   ###
   all_tables = liquibase_snapshot["snapshot"]["objects"]["liquibase.structure.core.Table"]
   
   ###
   ### Retrive maximum size from check definition
   ###
   max_index = int(liquibase_utilities.get_arg("MAX_INDEX"))
   
   ###
   ### Retrieve all changes in changeset
   ###
   changes = liquibase_utilities.get_changeset().getChanges()
   
   ###
   ### Loop through all changes
   ###
   for change in changes:
       ###
       ### LoadData change types are not currently supported
       ###
       if "loaddatachange" in change.getClass().getSimpleName().lower():
           liquibase_logger.info("LoadData change type not supported. Statement skipped.")
           continue
       ###
       ### Retrieve sql as string, remove extra whitespace
       ###
       raw_sql = liquibase_utilities.strip_comments(liquibase_utilities.generate_sql(change)).casefold()
       raw_sql = " ".join(raw_sql.split())
       ###
       ### Split raw_sql into statements
       ###
       raw_statements = liquibase_utilities.split_statements(raw_sql)
       for raw_statement in raw_statements:
           ###
           ### Split raw_statement into list
           ###
           sql_list = raw_statement.split()
           ###
           ### CREATE [UNIQUE] INDEX NAME ON [SCHEMA.]TABLE (column1, column2, ...)
           ###
           try:
               if (sql_list[0] == "create") and (sql_list[1] == "index" or sql_list[2] == "index"):
                   start = sql_list.index("on")
                   ###
                   ### Remove schema and parenthesis if provided
                   ###
                   table_name = sql_list[start + 1].split(".")[-1]
                   start = table_name.rfind("(")
                   if start != -1:
                       table_name = table_name[0:start]
               else:
                   raise UserWarning
           except (IndexError, ValueError):
               liquibase_logger.warning(f"Unsupported Create Index statement skipped: {raw_statement}")
               continue
           except UserWarning:
               liquibase_logger.info(f"Non Create Index statement skipped: {raw_statement}")
               continue
           ###
           ### Locate table
           ###
           table_object = find_snapshot_object(all_tables, "table", "name", table_name.strip())
           if table_object is None:
               liquibase_logger.warning(f"Table \"{table_name}\" not found in snapshot. Statement skipped.")
               continue
           table_name = table_object['table']['name']
           ###
           ### Sum indexes, check for maximum
           ###
           index_total = liquibase_utilities.get_cache(table_name, 1)
           if "indexes" in table_object["table"]:
               index_total += len(table_object["table"]["indexes"])
           liquibase_utilities.put_cache(table_name, index_total)
           if index_total > max_index:
               liquibase_status.fired = True
               status_message = str(liquibase_utilities.get_script_message()).replace("__TABLE_NAME__", f"\"{table_name}\"")
               status_message = status_message.replace("__INDEX_COUNT__", str(index_total))
               liquibase_status.message = status_message
               sys.exit(1)
   
   ###
   ### Default return code
   ###
   False

3. Initiate the customization process. In the CLI, run this command:
   liquibase checks customize --check-name=CustomCheckTemplate
   The CLI prompts you to finish configuring your file. A message displays:

   This check cannot be customized directly because one or more fields does not have a default value.

   Liquibase will then create a copy of CustomCheckTemplate and initiate the customization workflow.

4. Give your check a short name so you can easily identify what Python script it is associated with (up to 64 alpha-numeric characters only).
   In this example we will name the check:
   createIndexCount

5. Set the Severity to return a code of 0-4 when triggered. These severity codes allow you to determine if the job moves forward or stops when this check triggers.
   Learn more here: Use Policy Checks in Automation: Severity and Exit Code
   options: 'INFO'=0, 'MINOR'=1, 'MAJOR'=2, 'CRITICAL'=3, 'BLOCKER'=4

6. Set SCRIPT_DESCRIPTION. In this example, we will set the description to:

   This script ensures a table has less than a specified number of indexes.

7. Set SCRIPT_SCOPE. In this example, we will set the scope to changelog. The Python sample provided in this tutorial requires it.
   changelog: for example, if your check looks for syntax patterns or attributes in your Liquibase Changelog (the changes you author in your repository). With this value, the check runs once per changeset.
8. Set the SCRIPT_MESSAGE. This message will display when the check is triggered. In this example we will use:
   This script identified that Table __TABLE_NAME__ has __INDEX_COUNT__ indexes.
9. Set the SCRIPT_PATH. This is the relative path where your script is stored in relation to the changelog specified in --changelog-file, whether it is stored locally or in a repository.
   In this example, we will set the path to: scripts/create-index-count.py.
10. Set the SCRIPT_ARGUMENT. This allows you to pass dynamic information into the custom policy check without modifying the Python code.
    
    Specify MAX_SIZE=2 in the CLI to retrieve it in your code with the variable: max_size = int (liquibase_utilities.get_arg("MAX_SIZE")). If you customize your check later, you can specify a new value in the CLI. If you don't need dynamic arguments, leave this field blank.
11. Set the REQUIRES_SNAPSHOT. If your script scope is changelog, set whether the check requires a database snapshot. Specify true if your check needs to inspect database objects. (If your script scope is database, Liquibase always takes a snapshot, so this prompt does not appear.)

Note: The larger your database, the more performance impact a snapshot causes. If you cannot run a snapshot due to memory limitations, see Memory Limits of Inspecting Large Schemas.

You have now successfully created and customized a policy check!

Next steps

• Run a Custom Policy Check

Related Topics

Sample Custom Policy Check Scripts