blob: f57ccbb7a0291f5384ba4abc03831161bbf61670 (
plain)
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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
|
from django.apps import AppConfig
from django.db import connections
class BaseAppConfig(AppConfig):
name = "apps.query_performing_app"
database = "default"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.query_results = []
def ready(self):
self.query_results = []
self._perform_query()
def _perform_query(self):
raise NotImplementedError
class ModelQueryAppConfig(BaseAppConfig):
def _perform_query(self):
from ..models import TotallyNormal
queryset = TotallyNormal.objects.using(self.database)
queryset.update_or_create(name="new name")
self.query_results = list(queryset.values_list("name"))
class QueryDefaultDatabaseModelAppConfig(ModelQueryAppConfig):
database = "default"
class QueryOtherDatabaseModelAppConfig(ModelQueryAppConfig):
database = "other"
class CursorQueryAppConfig(BaseAppConfig):
def _perform_query(self):
connection = connections[self.database]
with connection.cursor() as cursor:
cursor.execute("SELECT 42" + connection.features.bare_select_suffix)
self.query_results = cursor.fetchall()
class QueryDefaultDatabaseCursorAppConfig(CursorQueryAppConfig):
database = "default"
class QueryOtherDatabaseCursorAppConfig(CursorQueryAppConfig):
database = "other"
class CursorQueryManyAppConfig(BaseAppConfig):
def _perform_query(self):
from ..models import TotallyNormal
connection = connections[self.database]
table_meta = TotallyNormal._meta
with connection.cursor() as cursor:
cursor.executemany(
"INSERT INTO %s (%s) VALUES(%%s)"
% (
connection.introspection.identifier_converter(table_meta.db_table),
connection.ops.quote_name(table_meta.get_field("name").column),
),
[("test name 1",), ("test name 2",)],
)
self.query_results = []
class QueryDefaultDatabaseCursorManyAppConfig(CursorQueryManyAppConfig):
database = "default"
class QueryOtherDatabaseCursorManyAppConfig(CursorQueryManyAppConfig):
database = "other"
class StoredProcedureQueryAppConfig(BaseAppConfig):
def _perform_query(self):
with connections[self.database].cursor() as cursor:
cursor.callproc("test_procedure")
self.query_results = []
class QueryDefaultDatabaseStoredProcedureAppConfig(StoredProcedureQueryAppConfig):
database = "default"
class QueryOtherDatabaseStoredProcedureAppConfig(StoredProcedureQueryAppConfig):
database = "other"
|
