1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
|
import json
from django import forms
from django.core.exceptions import ValidationError
from django.db import connections
from django.utils.functional import cached_property
from debug_toolbar.panels.sql.utils import is_select_query, reformat_sql
class SQLSelectForm(forms.Form):
"""
Validate params
sql: The sql statement with interpolated params
raw_sql: The sql statement with placeholders
params: JSON encoded parameter values
duration: time for SQL to execute passed in from toolbar just for redisplay
"""
sql = forms.CharField()
raw_sql = forms.CharField()
params = forms.CharField()
alias = forms.CharField(required=False, initial="default")
duration = forms.FloatField()
def clean_raw_sql(self):
value = self.cleaned_data["raw_sql"]
if not is_select_query(value):
raise ValidationError("Only 'select' queries are allowed.")
return value
def clean_params(self):
value = self.cleaned_data["params"]
try:
return json.loads(value)
except ValueError as exc:
raise ValidationError("Is not valid JSON") from exc
def clean_alias(self):
value = self.cleaned_data["alias"]
if value not in connections:
raise ValidationError(f"Database alias '{value}' not found")
return value
def reformat_sql(self):
return reformat_sql(self.cleaned_data["sql"], with_toggle=False)
@property
def connection(self):
return connections[self.cleaned_data["alias"]]
@cached_property
def cursor(self):
return self.connection.cursor()
|
