demo_validation.pyā¢3.24 kB
#!/usr/bin/env python3
"""
Demo script for PostgreSQL MCP Server Query Validation
This script demonstrates the query validation and optimization features
without requiring a database connection.
"""
import asyncio
from unittest.mock import MagicMock, AsyncMock
from src.query_optimizer import QueryValidator
async def demo_query_validation():
"""Demonstrate query validation capabilities"""
# Mock database manager (no real connection needed for validation demo)
mock_db_manager = MagicMock()
mock_db_manager.execute_query = AsyncMock()
mock_db_manager.get_column_info = AsyncMock(return_value=[])
# Create validator
validator = QueryValidator(mock_db_manager)
print("š§ PostgreSQL MCP Server - Query Validation Demo")
print("=" * 60)
print()
# Test queries with different characteristics
test_queries = [
{
"name": "ā
Good Query",
"sql": "SELECT id, name, email FROM users WHERE active = true ORDER BY created_at DESC LIMIT 10;"
},
{
"name": "ā ļø Performance Issues",
"sql": "SELECT * FROM users WHERE email LIKE '%@gmail.com' ORDER BY created_at;"
},
{
"name": "š Security Risk",
"sql": "SELECT * FROM users WHERE id = 1; DROP TABLE users; --"
},
{
"name": "ā Dangerous Operation",
"sql": "DROP TABLE users;"
},
{
"name": "š§ Complex Query",
"sql": """
SELECT u.name, COUNT(p.id) as post_count, AVG(c.rating) as avg_rating
FROM users u
LEFT JOIN posts p ON u.id = p.user_id
LEFT JOIN comments c ON p.id = c.post_id
WHERE u.created_at > '2023-01-01'
GROUP BY u.id, u.name
HAVING COUNT(p.id) > 5
ORDER BY post_count DESC, avg_rating DESC
"""
}
]
for i, test_case in enumerate(test_queries, 1):
print(f"š Test {i}: {test_case['name']}")
print("-" * 40)
try:
# Validate the query
analysis = await validator.validate_query(test_case['sql'].strip())
# Format and display results
report = validator.format_analysis_report(analysis)
print(report)
except Exception as e:
print(f"ā Validation error: {e}")
print()
print("=" * 60)
print()
print("šÆ Key Features Demonstrated:")
print("ā¢ SQL injection detection and prevention")
print("ā¢ Performance issue identification")
print("ā¢ Query complexity analysis")
print("ā¢ Optimization suggestions")
print("ā¢ Security validation")
print()
print("š” Try running these queries through the 'validate_query' tool!")
def main():
"""Main demo function"""
try:
asyncio.run(demo_query_validation())
except KeyboardInterrupt:
print("\nš Demo interrupted by user")
except ImportError as e:
print(f"ā Missing dependencies: {e}")
print("Install with: pip install -r requirements.txt")
except Exception as e:
print(f"ā Demo failed: {e}")
if __name__ == "__main__":
main()