Created
June 24, 2025 15:19
-
-
Save ariesmaulana/654ec4a553ddc15189cc50f4ac85f25b to your computer and use it in GitHub Desktop.
simple django orm profiling
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# orm_tester.py | |
# code generated by claude 4 | |
import time | |
import tracemalloc | |
from functools import wraps | |
from django.db import connection | |
class ORMTester: | |
"""Simple database-agnostic ORM tester""" | |
def __init__(self): | |
self.reset_queries() | |
def reset_queries(self): | |
"""Reset the connection queries log""" | |
connection.queries_log.clear() | |
def get_last_query(self): | |
"""Get the last executed SQL query""" | |
if connection.queries: | |
return connection.queries[-1]['sql'] | |
return "No query executed" | |
def format_memory(self, bytes_mem): | |
"""Format memory usage""" | |
if bytes_mem < 1024 * 1024: | |
return f"{bytes_mem / 1024:.2f} KB" | |
else: | |
return f"{bytes_mem / (1024 * 1024):.2f} MB" | |
def test_orm_method(self, func): | |
"""Decorator to test ORM methods""" | |
@wraps(func) | |
def wrapper(*args, **kwargs): | |
# Start memory tracking | |
tracemalloc.start() | |
# Reset queries log | |
self.reset_queries() | |
# Record start time | |
start_time = time.perf_counter() | |
# Execute the function | |
result = func(*args, **kwargs) | |
# Force evaluation if it's a QuerySet | |
if hasattr(result, '__iter__') and hasattr(result, 'query'): | |
# Convert to list and access related fields to trigger all queries | |
evaluated_result = list(result) | |
# For select_related/prefetch_related, access related fields | |
if evaluated_result: | |
first_obj = evaluated_result[0] | |
# Try to access all related fields to trigger any lazy loading | |
for field in first_obj._meta.get_fields(): | |
if hasattr(field, 'related_model') and field.related_model: | |
try: | |
# Access the related field to trigger query if needed | |
getattr(first_obj, field.name, None) | |
except: | |
pass | |
elif hasattr(field, 'remote_field') and field.remote_field: | |
try: | |
# For reverse relations | |
if hasattr(first_obj, field.get_accessor_name()): | |
related_manager = getattr(first_obj, field.get_accessor_name()) | |
if hasattr(related_manager, 'all'): | |
list(related_manager.all()) | |
except: | |
pass | |
# Record end time | |
end_time = time.perf_counter() | |
# Get memory usage | |
current, peak = tracemalloc.get_traced_memory() | |
tracemalloc.stop() | |
# Get execution time in milliseconds | |
exec_time = (end_time - start_time) * 1000 | |
# Get all executed queries (important for select_related/prefetch_related) | |
all_queries = [q['sql'] for q in connection.queries] | |
# Print results | |
if len(all_queries) == 1: | |
print(f"Query Generated: {all_queries[0]}") | |
else: | |
print(f"Queries Generated ({len(all_queries)} total):") | |
for i, query in enumerate(all_queries, 1): | |
print(f" {i}. {query}") | |
if i > 3: | |
break | |
print(f"Mem Usage: {self.format_memory(peak)}") | |
print(f"Exec Time: {exec_time:.2f} ms") | |
# Return None to avoid QuerySet output pollution | |
return None | |
return wrapper | |
# Create default tester instance | |
orm_tester = ORMTester() | |
# Convenience decorator | |
def test_orm_method(func): | |
"""Simple decorator for testing ORM methods""" | |
return orm_tester.test_orm_method(func) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
how to use