collectionMustHaveValidator

Last updated: September 2, 2025

collectionMustHaveValidator is a custom policy check that checks for a validator when creating a new collection.

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

This example utilizes MongoDB. You can use this check as it is or customize it further to fit your needs in your NoSQL database.

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

Before you begin

Scope

Database

database

MongoDB

Before you begin

Procedure

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

1

Add this code to your Checks Settings file:

collectionMustHaveValidator Python Script
# # #
# # # This script checks
for a validator when creating a new collection
# # #

# # #
# # # Helpers come from Liquibase
# # #
import sys
import liquibase_database
import liquibase_utilities

# # #
# # # Constants
# # #
NOSQL_DATABASES = ["MongoDB"]

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

# # #
# # # Retrieve status handler
# # #
liquibase_status = liquibase_utilities.get_status()

# # #
# # # Check
for Mongo
# # #
current_database = liquibase_utilities.get_database()
product_name = liquibase_database.get_short_name(current_database)
if not product_name.casefold() in map(str.casefold, NOSQL_DATABASES):
    liquibase_logger.info(f "Database {product_name} ignored")
liquibase_status.fired = False
sys.exit(1)

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

# # #
# # # Loop through all changes
# # #
for change in changes:
    # # #
# # # 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())

# # #
# # # Look
for createCollection
# # #
if "createcollection" in raw_sql and not "validator:" in raw_sql:
    liquibase_status.fired = True
liquibase_status.message = liquibase_utilities.get_script_message()
sys.exit(1)

# # #
# # # Default
return code
# # #
False
2

Initiate the customization process

A short name is a descriptive name that indicates what you want the check to search for in the database or changelog. In this example, we will provideSqlTruncateCheckas the short name, because we will be searching for any instances ofTruncate. The CLI will indicate that the new check was created fromSqlUserDefinedPatterncheck successfully:

New check 'SqlTruncateCheck' created from 'SqlUserDefinedPatternCheck'

3

Give your check a short name so you can easily identify what Python script it is associated with

You may use up to 64 alpha-numeric characters only.

In this example we will name the check: collectionMustHaveValidator

4

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

5

Set SCRIPT_DESCRIPTION

This script checks for a validator when creating a new collection.
6

Set the SCRIPT_SCOPE

In this example, we will set the scope to:

  • database: If your check looks for the presence of keys, indexes, or table name patterns in your database schema including Liquibase Tracking Tables. With this value, the check runs once for each database object.

7

Set the SCRIPT_MESSAGE

This message will display when the check is triggered. In this example we will use:

collectionMustHaveValidator Python Script
# # #
# # # This script checks
for a validator when creating a new collection
# # #

# # #
# # # Helpers come from Liquibase
# # #
import sys
import liquibase_database
import liquibase_utilities

# # #
# # # Constants
# # #
NOSQL_DATABASES = ["MongoDB"]

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

# # #
# # # Retrieve status handler
# # #
liquibase_status = liquibase_utilities.get_status()

# # #
# # # Check
for Mongo
# # #
current_database = liquibase_utilities.get_database()
product_name = liquibase_database.get_short_name(current_database)
if not product_name.casefold() in map(str.casefold, NOSQL_DATABASES):
    liquibase_logger.info(f "Database {product_name} ignored")
liquibase_status.fired = False
sys.exit(1)

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

# # #
# # # Loop through all changes
# # #
for change in changes:
    # # #
# # # 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())

# # #
# # # Look
for createCollection
# # #
if "createcollection" in raw_sql and not "validator:" in raw_sql:
    liquibase_status.fired = True
liquibase_status.message = liquibase_utilities.get_script_message()
sys.exit(1)

# # #
# # # Default
return code
# # #
False
8

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/collection-camel-case.py

9

This check does not require a SCRIPT_ARGUMENT, so leave this blank.

10

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.

collectionMustHaveValidator - Liquibase