-
Notifications
You must be signed in to change notification settings - Fork 1.1k
Binary parameter handling for GeoDjango #2169
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
db28e81
5205a50
f8e2f67
f3ac881
a622823
8555739
457879b
0615f56
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,26 @@ | ||
import base64 | ||
import json | ||
|
||
|
||
class DebugToolbarJSONDecoder(json.JSONDecoder): | ||
"""Custom JSON decoder that reconstructs binary data during parsing.""" | ||
|
||
def decode(self, s): | ||
"""Override decode to apply reconstruction after parsing.""" | ||
obj = super().decode(s) | ||
return self._reconstruct_params(obj) | ||
|
||
def _reconstruct_params(self, params): | ||
"""Reconstruct parameters, handling lists and dicts recursively.""" | ||
if isinstance(params, list): | ||
return [self._reconstruct_params(param) for param in params] | ||
elif isinstance(params, dict): | ||
if "__djdt_binary__" in params: | ||
return base64.b64decode(params["__djdt_binary__"]) | ||
else: | ||
return { | ||
key: self._reconstruct_params(value) | ||
for key, value in params.items() | ||
} | ||
else: | ||
return params | ||
tim-schilling marked this conversation as resolved.
Show resolved
Hide resolved
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,3 +1,4 @@ | ||
import base64 | ||
import contextlib | ||
import contextvars | ||
import datetime | ||
|
@@ -126,6 +127,11 @@ def _decode(self, param): | |
if isinstance(param, dict): | ||
return {key: self._decode(value) for key, value in param.items()} | ||
|
||
# Handle binary data (e.g., GeoDjango EWKB geometry data) | ||
if isinstance(param, (bytes, bytearray)): | ||
# Mark as binary data for later reconstruction | ||
return {"__djdt_binary__": base64.b64encode(param).decode("ascii")} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We should move this logic into a shared utility that way we don't have the magic value I'm also curious if we should be using this with the Store logic. If we do, then we could potentially ignore this and handle it within |
||
|
||
# make sure datetime, date and time are converted to string by force_str | ||
CONVERT_TYPES = (datetime.datetime, datetime.date, datetime.time) | ||
try: | ||
|
Original file line number | Diff line number | Diff line change | ||||||||
---|---|---|---|---|---|---|---|---|---|---|
@@ -0,0 +1,137 @@ | ||||||||||
""" | ||||||||||
Tests for GeoDjango binary parameter handling fix | ||||||||||
""" | ||||||||||
|
||||||||||
import base64 | ||||||||||
import json | ||||||||||
|
||||||||||
from debug_toolbar.panels.sql.decoders import DebugToolbarJSONDecoder | ||||||||||
from debug_toolbar.panels.sql.tracking import NormalCursorMixin | ||||||||||
|
||||||||||
from ..base import BaseTestCase | ||||||||||
|
||||||||||
|
||||||||||
class MockCursor: | ||||||||||
"""Mock cursor for testing""" | ||||||||||
|
||||||||||
|
||||||||||
class MockConnection: | ||||||||||
"""Mock database connection for testing""" | ||||||||||
|
||||||||||
vendor = "postgresql" | ||||||||||
alias = "default" | ||||||||||
|
||||||||||
|
||||||||||
class MockLogger: | ||||||||||
"""Mock logger for testing""" | ||||||||||
|
||||||||||
def record(self, **kwargs): | ||||||||||
pass | ||||||||||
|
||||||||||
|
||||||||||
class StubCursor(NormalCursorMixin): | ||||||||||
"""Test cursor that can be instantiated""" | ||||||||||
|
||||||||||
def __init__(self): | ||||||||||
self.cursor = MockCursor() | ||||||||||
self.db = MockConnection() | ||||||||||
self.logger = MockLogger() | ||||||||||
|
||||||||||
|
||||||||||
class GeoDjangoBinaryParameterTest(BaseTestCase): | ||||||||||
"""Test cases for GeoDjango binary parameter handling""" | ||||||||||
|
||||||||||
def test_binary_parameter_encoding_decoding(self): | ||||||||||
"""Test that binary parameters are properly encoded and decoded""" | ||||||||||
cursor = TestCursor() | ||||||||||
|
||||||||||
# Test binary data similar to GeoDjango EWKB geometry | ||||||||||
binary_data = b"\x01\x01\x00\x00\x20\xe6\x10\x00\x00\xff\xfe\xfd" | ||||||||||
encoded = cursor._decode(binary_data) | ||||||||||
|
||||||||||
self.assertIsInstance(encoded, dict) | ||||||||||
self.assertIn("__djdt_binary__", encoded) | ||||||||||
|
||||||||||
expected_b64 = base64.b64encode(binary_data).decode("ascii") | ||||||||||
self.assertEqual(encoded["__djdt_binary__"], expected_b64) | ||||||||||
|
||||||||||
json_params = json.dumps([encoded]) | ||||||||||
reconstructed = json.loads(json_params, cls=DebugToolbarJSONDecoder) | ||||||||||
|
||||||||||
self.assertEqual(len(reconstructed), 1) | ||||||||||
self.assertEqual(reconstructed[0], binary_data) | ||||||||||
self.assertIsInstance(reconstructed[0], bytes) | ||||||||||
Comment on lines
+61
to
+63
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I don't think there's a benefit to being this verbose. Try to be more concise such as:
Suggested change
|
||||||||||
|
||||||||||
def test_mixed_parameter_types(self): | ||||||||||
"""Test that mixed parameter types are handled correctly""" | ||||||||||
cursor = TestCursor() | ||||||||||
|
||||||||||
params = [ | ||||||||||
"string_param", | ||||||||||
42, | ||||||||||
b"\x01\x02\x03", | ||||||||||
None, | ||||||||||
["nested", "list"], | ||||||||||
] | ||||||||||
|
||||||||||
encoded_params = [cursor._decode(p) for p in params] | ||||||||||
|
||||||||||
json_str = json.dumps(encoded_params) | ||||||||||
reconstructed = json.loads(json_str, cls=DebugToolbarJSONDecoder) | ||||||||||
|
||||||||||
self.assertEqual(reconstructed[0], "string_param") # string unchanged | ||||||||||
self.assertEqual(reconstructed[1], 42) # int unchanged | ||||||||||
self.assertEqual(reconstructed[2], b"\x01\x02\x03") # binary restored | ||||||||||
self.assertIsNone(reconstructed[3]) # None unchanged | ||||||||||
self.assertEqual(reconstructed[4], ["nested", "list"]) # list unchanged | ||||||||||
|
||||||||||
def test_nested_binary_data(self): | ||||||||||
"""Test binary data nested in lists and dicts""" | ||||||||||
cursor = TestCursor() | ||||||||||
|
||||||||||
nested_params = [ | ||||||||||
[b"\x01\x02", "string", b"\x03\x04"], | ||||||||||
{"key": b"\x05\x06", "other": "value"}, | ||||||||||
] | ||||||||||
|
||||||||||
encoded = [cursor._decode(p) for p in nested_params] | ||||||||||
|
||||||||||
json_str = json.dumps(encoded) | ||||||||||
reconstructed = json.loads(json_str, cls=DebugToolbarJSONDecoder) | ||||||||||
|
||||||||||
self.assertEqual(reconstructed[0][0], b"\x01\x02") | ||||||||||
self.assertEqual(reconstructed[0][1], "string") | ||||||||||
self.assertEqual(reconstructed[0][2], b"\x03\x04") | ||||||||||
|
||||||||||
self.assertEqual(reconstructed[1]["key"], b"\x05\x06") | ||||||||||
self.assertEqual(reconstructed[1]["other"], "value") | ||||||||||
|
||||||||||
def test_empty_binary_data(self): | ||||||||||
"""Test handling of empty binary data""" | ||||||||||
cursor = TestCursor() | ||||||||||
|
||||||||||
empty_bytes = b"" | ||||||||||
encoded = cursor._decode(empty_bytes) | ||||||||||
|
||||||||||
self.assertIsInstance(encoded, dict) | ||||||||||
self.assertIn("__djdt_binary__", encoded) | ||||||||||
|
||||||||||
json_str = json.dumps([encoded]) | ||||||||||
reconstructed = json.loads(json_str, cls=DebugToolbarJSONDecoder) | ||||||||||
|
||||||||||
self.assertEqual(reconstructed[0], empty_bytes) | ||||||||||
|
||||||||||
def test_bytearray_support(self): | ||||||||||
"""Test that bytearray is also handled as binary data""" | ||||||||||
cursor = TestCursor() | ||||||||||
|
||||||||||
byte_array = bytearray(b"\x01\x02\x03\x04") | ||||||||||
encoded = cursor._decode(byte_array) | ||||||||||
|
||||||||||
self.assertIn("__djdt_binary__", encoded) | ||||||||||
|
||||||||||
json_str = json.dumps([encoded]) | ||||||||||
reconstructed = json.loads(json_str, cls=DebugToolbarJSONDecoder) | ||||||||||
|
||||||||||
self.assertEqual(reconstructed[0], byte_array) | ||||||||||
self.assertIsInstance(reconstructed[0], bytes) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Rather than the approach you've taken here, you may want to consider going with something like:
This will require you to change the tests. The decoding will only work if there's an actual object being passed in. Most of the tests are passing in a list of values. Using an object/dictionary is more representative of what will actually be used since this is for parameters for a SQL query which is a dictionary.