Pony Object-Relational Mapper for Python with Pythonic query syntax using generator expressions
—
Tools for debugging SQL queries, inspecting schemas, and configuring runtime behavior. Essential for development and troubleshooting database operations in Pony ORM applications.
Functions for monitoring and debugging generated SQL queries and their execution.
def set_sql_debug(debug=True, show_values=None):
"""Configure SQL debugging with fine-grained control.
Args:
debug: Enable/disable SQL query logging (default: True)
show_values: Show parameter values in debug output (default: None)
- True: Always show values
- False: Never show values
- None: Show values based on debug level
Usage:
set_sql_debug(True) # Enable SQL logging
set_sql_debug(True, show_values=True) # Show SQL with parameters
set_sql_debug(False) # Disable SQL logging
"""
def sql_debug(value):
"""Legacy function for enabling/disabling SQL debugging.
Args:
value: Boolean to enable/disable SQL debugging
Note: Deprecated - use set_sql_debug() instead for better control
"""
class sql_debugging:
"""Context manager for temporary SQL debugging activation.
Usage:
with sql_debugging:
# SQL queries will be printed during this block
users = select(u for u in User if u.active)
"""
def __enter__(self):
"""Enter debugging context."""
def __exit__(self, exc_type, exc_val, exc_tb):
"""Exit debugging context and restore previous setting."""Functions for inspecting database schemas and pretty-printing query results.
def show(entity_or_query):
"""Pretty-print entity schema or query results for debugging.
Args:
entity_or_query: Entity class, entity instance, or query object
Usage:
show(User) # Display User entity schema
show(user_instance) # Display entity instance data
show(select(u for u in User)) # Display query results
"""from pony.orm import *
db = Database()
class User(db.Entity):
name = Required(str)
email = Required(str, unique=True)
age = Optional(int)
db.bind('sqlite', filename='debug_example.db')
db.generate_mapping(create_tables=True)
# Enable SQL debugging globally
set_sql_debug(True)
with db_session:
# All SQL queries will now be printed
user = User(name="Alice", email="alice@example.com", age=25)
# Prints: INSERT INTO User (name, email, age) VALUES (?, ?, ?)
users = select(u for u in User if u.age > 18)
# Prints: SELECT u.id, u.name, u.email, u.age FROM User u WHERE u.age > ?
adult_count = count(u for u in User if u.age >= 18)
# Prints: SELECT COUNT(*) FROM User u WHERE u.age >= ?
# Disable SQL debugging
set_sql_debug(False)
with db_session:
# No SQL output
more_users = select(u for u in User)# Enable debugging with parameter values
set_sql_debug(True, show_values=True)
with db_session:
# Query with parameters - values will be shown
young_users = select(u for u in User if u.age < 30)
# Prints: SELECT u.id, u.name, u.email, u.age FROM User u WHERE u.age < ?
# Parameters: [30]
user = User.get(email="alice@example.com")
# Prints: SELECT u.id, u.name, u.email, u.age FROM User u WHERE u.email = ?
# Parameters: ['alice@example.com']
# Complex query with multiple parameters
filtered_users = select(u for u in User
if u.age >= 18 and u.name.startswith("A"))
# Shows full SQL with all parameter values
# Hide parameter values for security
set_sql_debug(True, show_values=False)
with db_session:
# SQL shown but parameters hidden
user = User.get(email="sensitive@example.com")
# Prints: SELECT u.id, u.name, u.email, u.age FROM User u WHERE u.email = ?
# (no parameter values shown)# Normal operations without debug output
with db_session:
user1 = User(name="Bob", email="bob@example.com")
# Temporary debugging for specific operations
with sql_debugging:
with db_session:
# Only these queries will show SQL
user2 = User(name="Charlie", email="charlie@example.com")
users = select(u for u in User if u.name.startswith("C"))
# Complex query to debug
result = select((u.name, u.age) for u in User
if u.age and u.age > 20).order_by(u.age)
# Back to normal (no debug output)
with db_session:
user3 = User(name="Dave", email="dave@example.com")# Display entity schema
show(User)
# Output:
# class User(Entity):
# id = PrimaryKey(int, auto=True)
# name = Required(str)
# email = Required(str, unique=True)
# age = Optional(int)
with db_session:
# Create some test data
users = [
User(name="Alice", email="alice@example.com", age=25),
User(name="Bob", email="bob@example.com", age=30),
User(name="Charlie", email="charlie@example.com", age=22)
]
# Display entity instance
show(users[0])
# Output:
# User[1]
# id: 1
# name: 'Alice'
# email: 'alice@example.com'
# age: 25
# Display query results
query = select(u for u in User if u.age >= 25)
show(query)
# Output:
# User[1]
# User[2]
# (shows all matching users)
# Pretty print query with projections
name_age_query = select((u.name, u.age) for u in User).order_by(u.name)
show(name_age_query)
# Output:
# name | age
# ---------|----
# Alice | 25
# Bob | 30
# Charlie | 22with db_session:
# Get query object without executing
query = select(u for u in User if u.age > 20)
# Show SQL for specific query
query.show()
# Prints the SQL for this specific query
# Get SQL string programmatically
sql_string = query.get_sql()
print(f"Generated SQL: {sql_string}")
# Execute query after inspection
results = list(query)
# Debug complex queries step by step
with db_session:
base_query = select(u for u in User)
print("Base query:")
base_query.show()
filtered_query = base_query.filter(lambda u: u.age > 18)
print("After age filter:")
filtered_query.show()
ordered_query = filtered_query.order_by(User.name)
print("After ordering:")
ordered_query.show()
final_query = ordered_query.limit(10)
print("After limit:")
final_query.show()import os
import logging
# Environment-based debugging configuration
def setup_pony_debugging():
debug_level = os.getenv('PONY_DEBUG', 'off').lower()
if debug_level == 'full':
# Full debugging with parameter values
set_sql_debug(True, show_values=True)
elif debug_level == 'sql':
# SQL only without parameters
set_sql_debug(True, show_values=False)
elif debug_level == 'off':
# No debugging
set_sql_debug(False)
return debug_level
# Setup debugging based on environment
debug_mode = setup_pony_debugging()
print(f"Pony debugging mode: {debug_mode}")
# Custom logging for SQL queries
class SQLLogger:
def __init__(self, logger_name='pony.sql'):
self.logger = logging.getLogger(logger_name)
def log_query(self, sql, params=None):
if params:
self.logger.info(f"SQL: {sql} | Params: {params}")
else:
self.logger.info(f"SQL: {sql}")
# Production debugging with custom logger
if os.getenv('ENVIRONMENT') == 'production':
# Use logging instead of print for production
logging.basicConfig(level=logging.INFO)
sql_logger = SQLLogger()
# Custom query execution with logging
def logged_query(query_func):
with sql_debugging:
# Capture SQL output and log it properly
result = query_func()
return result
with db_session:
users = logged_query(lambda: list(select(u for u in User)))import time
from contextlib import contextmanager
@contextmanager
def query_timer(query_name="Query"):
"""Time query execution for performance debugging."""
start_time = time.time()
try:
yield
finally:
end_time = time.time()
duration = end_time - start_time
print(f"{query_name} took {duration:.4f} seconds")
# Combine SQL debugging with timing
with sql_debugging:
with db_session:
with query_timer("User count"):
user_count = count(u for u in User)
with query_timer("Complex aggregation"):
stats = select(
(count(), avg(u.age), min(u.age), max(u.age))
for u in User if u.age
).get()
with query_timer("Join query"):
user_orders = select(
(u.name, count(o.id))
for u in User for o in u.orders
).group_by(u.id)
# Profile multiple queries
def profile_queries():
queries = {
"all_users": lambda: list(select(u for u in User)),
"adult_users": lambda: list(select(u for u in User if u.age >= 18)),
"user_count": lambda: count(u for u in User),
"avg_age": lambda: avg(u.age for u in User if u.age)
}
with sql_debugging:
with db_session:
for name, query_func in queries.items():
with query_timer(name):
result = query_func()
print(f" Result: {result}")
# Run profiling
profile_queries()# Good: Enable debugging for development
if __name__ == "__main__":
set_sql_debug(True, show_values=True)
# Good: Conditional debugging based on environment
import logging
if logging.getLogger().getEffectiveLevel() == logging.DEBUG:
set_sql_debug(True)
# Good: Temporary debugging for specific issues
def debug_user_creation():
with sql_debugging:
with db_session:
try:
user = User(name="Test", email="test@example.com")
commit()
except Exception as e:
print(f"Error creating user: {e}")
show(User) # Inspect schema
raise
# Good: Debug query performance issues
def debug_slow_query():
set_sql_debug(True, show_values=True)
with db_session:
print("=== Debugging slow query ===")
# Original slow query
slow_query = select(u for u in User
if exists(o for o in u.orders
if o.total > 1000))
print("Original query:")
slow_query.show()
# Optimized version
fast_query = select(u for u in User
if u.id in select(o.user_id for o in Order
if o.total > 1000))
print("Optimized query:")
fast_query.show()
set_sql_debug(False)
# Bad: Leave debugging enabled in production
# set_sql_debug(True) # Don't do this in production code
# Bad: Debug without context
# with db_session:
# users = select(u for u in User) # Hard to identify which query
# show(users)