Sample Custom Policy Check Scripts

This page provides executable sample Python scripts that you can run as custom policy checks in Liquibase. These samples are representative of how you can structure your own scripts.

For even more samples, see this GitHub repository: liquibase/custom_policychecks. Liquibase provides a number of real-world samples of custom policy checks in this repository.

Guidance on writing Python code

Warning: Custom policy checks are not isolated and can interact both with local file systems and network utilities like the targeted database. It is a best practice to review all checks prior to execution to ensure they only affect the intended object(s).

Liquibase modules

When you download the Liquibase Checks extension, you also get a built-in set of Liquibase modules that you can use to easily call the Liquibase API. For example, there are modules containing functions to:

• Find the ID of a changeset
• Get a list of columns in a table
• Check if a database object is a table, column, index, foreign key, or other object
• Modify internal Python script bindings

For a list of Liquibase modules and functions you can call on, see Liquibase Python Modules.

Python libraries

If you're new to Python, read the official Python tutorial before writing any scripts. See the Python Standard Library for information on built-in functions. You can always import these at the beginning of your script.

Liquibase comes pre-installed with a GraalPy virtual environment. In addition to the built-in Liquibase modules, this virtual environment contains the following Python modules that aren't part of the Python Standard Library:

• simplejson
• sqlparse
• urllib3

However, you cannot install additional modules in the built-in virtual environment because it is packaged in the Liquibase checks extension JAR file. To install additional modules, you must create your own virtual environment and install them yourself, including a package containing the Liquibase API. For more information, see Create a Python Virtual Environment.

Looping behavior

When you execute the checks run command, Liquibase always runs every enabled policy check against every object the check targets. This works the same for custom policy checks. For example, Liquibase runs any custom policy check with a --checks-scope setting of:

• changelog: once per changeset in that changelog. If the changelog uses include or includeAll to reference child changelogs, the script also runs once per changeset in each of those child changelogs.
• database: once per database object. Liquibase generates a database snapshot once and then runs your script against every object in the snapshot. Liquibase runs your check against the schema(s) you specify.

Note: If you write a custom policy check that is individually performance-heavy, and you have a very large changelog or database, Liquibase may take a long time to run your custom check against all objects.

Script status

Your Python script doesn't need to return a value. Liquibase uses an object called Status to track the state of your check:

• Whether the check has triggered (fired)
• The message to display as output

At the point in your script where the check should trigger, use the liquibase_utilities module to modify the Status object:

if [trigger condition]:
    liquibase_utilities.get_status().fired = True
    liquibase_utilities.get_status().message = liquibase_utilities.get_script_message()
    sys.exit(1)

False

Notes on the preceding code sample:

• liquibase_utilities.get_status().fired is set to False by default. You must set it to True or else the Liquibase policy checks framework assumes the check didn't trigger.
• liquibase_utilities.get_script_message() returns the static string you specify when you create your Liquibase custom policy check object in the CLI.
  • If you left the CLI field blank, get_script_message() returns an empty string. If you want to add a message later, you have to customize your check or specify your message directly in your Python script as a string.
  • If you want to insert dynamic content into this string, such as the name of a database object, see Write Dynamic Status Messages for Custom Policy Checks.
• Normally, sys.exit(1) signifies a program failure. However, in this case you just want to use it to force Python to exit. In this case, Liquibase is working just how you want it to! You can specify a more meaningful exit code by configuring the severity level of your custom check in the CLI. See Create a Custom Policy Check.
• Specify False at the end of your script to follow Python convention. This code is evaluated if the check doesn't trigger, but has no effect.

Sample scripts

# This script checks for the phrase "DELETE FROM" without "WHERE"

# Modules come from Liquibase, SQLParse, and Python
import liquibase_utilities
import sqlparse
import sys

# Retrieve log handler
# Ex. liquibase_logger.info(message)
liquibase_logger = liquibase_utilities.get_logger()

# Retrieve status handler
liquibase_status = liquibase_utilities.get_status()

# Retrieve all changes in changeset
changes = liquibase_utilities.get_changeset().getChanges()

# Loop through all changes (within a single changeset)
for change in changes:

    # LoadData change types are not currently supported
    if "loaddatachange" in change.getClass().getSimpleName().lower():
        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 sql into statements
    raw_statements = liquibase_utilities.split_statements(raw_sql)
    for raw_statement in raw_statements:

        # Get list of token objects, convert to string
        tokens = liquibase_utilities.tokenize(raw_statement)
        keywords = [str(token) for token in tokens if token.is_keyword or isinstance(token, sqlparse.sql.Where)]
        keywords = [keyword for keyword in " ".join(keywords).split()]

        # Look for delete
        if len(keywords) >= 2 and keywords[0] == "delete" and keywords[1] == "from" and "where" not in keywords:
            liquibase_status.fired = True
            liquibase_status.message = liquibase_utilities.get_script_message()
            sys.exit(1)

# Default return code
False

# This script ensures all VARCHAR columns are under a maximum size

# Modules come from Python and Liquibase
import sys
import liquibase_utilities

# Retrieve log handler
# Ex. liquibase_logger.info(message)
liquibase_logger = liquibase_utilities.get_logger()

# Retrieve status handler
liquibase_status = liquibase_utilities.get_status()

# Retrive maximum size from check definition
max_size = int(liquibase_utilities.get_arg("VARCHAR_MAX"))

# Retrieve database object
database_object = liquibase_utilities.get_database_object()

# Skip if not a varchar column
if "column" in database_object.getObjectTypeName().lower() and "varchar" in str(database_object.getType()).lower():
    column_name = database_object.getName()
    column_size = int(database_object.getType().getColumnSize())

    if column_size > max_size:
        liquibase_status.fired = True
        status_message = str(liquibase_utilities.get_script_message()).replace("__COLUMN_NAME__", f"'{column_name}'")
        status_message = status_message.replace("__COLUMN_SIZE__", f"{max_size}")
        liquibase_status.message = status_message
        sys.exit(1)

# Default return code
False

If you need help deploying a finished check, see Create a Custom Policy Check.

Related links

• GitHub: liquibase/custom_policychecks