PgSanity
PgSanity is a Python utility that checks the syntax of PostgreSQL SQL files. It operates by leveraging the `ecpg` command-line tool, which is part of the PostgreSQL client development tools. This allows `pgsanity` to use the exact same parser as PostgreSQL to identify SQL syntax errors, making it a reliable tool for quality assurance and testing of SQL scripts. The current version is 0.3.0, and it maintains a regular release cadence.
Warnings
- gotcha PgSanity is a wrapper around the `ecpg` command, which is part of the PostgreSQL client development tools. These tools (specifically `ecpg`) must be installed on the system and accessible in the system's PATH for `pgsanity` to function correctly. This is an external system dependency, not a Python package dependency.
- gotcha PgSanity is designed as a command-line utility rather than a Python library intended for direct programmatic import and manipulation within Python scripts. While it's a Python package, its primary interface is through executing the `pgsanity` command, typically via `subprocess` calls in Python or directly from the shell.
Install
-
pip install pgsanity
Quickstart
# Check a single SQL file
pip install pgsanity
# Create a dummy SQL file
with open('test.sql', 'w') as f:
f.write('SELECT 1 FROM my_table;\n')
f.write('INSERT INTO another_table (id) VALUES (10);\n')
import subprocess
# Run pgsanity on the file
result = subprocess.run(['pgsanity', 'test.sql'], capture_output=True, text=True)
print(f"Exit Code: {result.returncode}")
print(f"STDOUT:\n{result.stdout}")
print(f"STDERR:\n{result.stderr}")
# Example of invalid SQL
with open('invalid.sql', 'w') as f:
f.write('SELECT * FROM non_existent_table WHERE bad_syntax;\n')
result_invalid = subprocess.run(['pgsanity', 'invalid.sql'], capture_output=True, text=True)
print(f"\nExit Code (invalid): {result_invalid.returncode}")
print(f"STDOUT (invalid):\n{result_invalid.stdout}")
print(f"STDERR (invalid):\n{result_invalid.stderr}")
# Check SQL from stdin
sql_to_check = 'SELECT current_timestamp;'
result_stdin = subprocess.run(['pgsanity'], input=sql_to_check, capture_output=True, text=True)
print(f"\nExit Code (stdin): {result_stdin.returncode}")
print(f"STDOUT (stdin):\n{result_stdin.stdout}")
print(f"STDERR (stdin):\n{result_stdin.stderr}")