Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/db/backends/base/introspection.py: 30%
68 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1from collections import namedtuple
3# Structure returned by DatabaseIntrospection.get_table_list()
4TableInfo = namedtuple("TableInfo", ["name", "type"])
6# Structure returned by the DB-API cursor.description interface (PEP 249)
7FieldInfo = namedtuple(
8 "FieldInfo",
9 "name type_code display_size internal_size precision scale null_ok "
10 "default collation",
11)
14class BaseDatabaseIntrospection:
15 """Encapsulate backend-specific introspection utilities."""
17 data_types_reverse = {}
19 def __init__(self, connection):
20 self.connection = connection
22 def get_field_type(self, data_type, description):
23 """
24 Hook for a database backend to use the cursor description to
25 match a Django field type to a database column.
27 For Oracle, the column data_type on its own is insufficient to
28 distinguish between a FloatField and IntegerField, for example.
29 """
30 return self.data_types_reverse[data_type]
32 def identifier_converter(self, name):
33 """
34 Apply a conversion to the identifier for the purposes of comparison.
36 The default identifier converter is for case sensitive comparison.
37 """
38 return name
40 def table_names(self, cursor=None, include_views=False):
41 """
42 Return a list of names of all tables that exist in the database.
43 Sort the returned table list by Python's default sorting. Do NOT use
44 the database's ORDER BY here to avoid subtle differences in sorting
45 order between databases.
46 """
48 def get_names(cursor):
49 return sorted(
50 ti.name
51 for ti in self.get_table_list(cursor)
52 if include_views or ti.type == "t"
53 )
55 if cursor is None: 55 ↛ 56line 55 didn't jump to line 56, because the condition on line 55 was never true
56 with self.connection.cursor() as cursor:
57 return get_names(cursor)
58 return get_names(cursor)
60 def get_table_list(self, cursor):
61 """
62 Return an unsorted list of TableInfo named tuples of all tables and
63 views that exist in the database.
64 """
65 raise NotImplementedError(
66 "subclasses of BaseDatabaseIntrospection may require a get_table_list() "
67 "method"
68 )
70 def get_table_description(self, cursor, table_name):
71 """
72 Return a description of the table with the DB-API cursor.description
73 interface.
74 """
75 raise NotImplementedError(
76 "subclasses of BaseDatabaseIntrospection may require a "
77 "get_table_description() method."
78 )
80 def get_migratable_models(self):
81 from django.apps import apps
82 from django.db import router
84 return (
85 model
86 for app_config in apps.get_app_configs()
87 for model in router.get_migratable_models(app_config, self.connection.alias)
88 if model._meta.can_migrate(self.connection)
89 )
91 def django_table_names(self, only_existing=False, include_views=True):
92 """
93 Return a list of all table names that have associated Django models and
94 are in INSTALLED_APPS.
96 If only_existing is True, include only the tables in the database.
97 """
98 tables = set()
99 for model in self.get_migratable_models():
100 if not model._meta.managed:
101 continue
102 tables.add(model._meta.db_table)
103 tables.update(
104 f.m2m_db_table()
105 for f in model._meta.local_many_to_many
106 if f.remote_field.through._meta.managed
107 )
108 tables = list(tables)
109 if only_existing:
110 existing_tables = set(self.table_names(include_views=include_views))
111 tables = [
112 t for t in tables if self.identifier_converter(t) in existing_tables
113 ]
114 return tables
116 def installed_models(self, tables):
117 """
118 Return a set of all models represented by the provided list of table
119 names.
120 """
121 tables = set(map(self.identifier_converter, tables))
122 return {
123 m
124 for m in self.get_migratable_models()
125 if self.identifier_converter(m._meta.db_table) in tables
126 }
128 def sequence_list(self):
129 """
130 Return a list of information about all DB sequences for all models in
131 all apps.
132 """
133 sequence_list = []
134 with self.connection.cursor() as cursor:
135 for model in self.get_migratable_models():
136 if not model._meta.managed:
137 continue
138 if model._meta.swapped:
139 continue
140 sequence_list.extend(
141 self.get_sequences(
142 cursor, model._meta.db_table, model._meta.local_fields
143 )
144 )
145 for f in model._meta.local_many_to_many:
146 # If this is an m2m using an intermediate table,
147 # we don't need to reset the sequence.
148 if f.remote_field.through._meta.auto_created:
149 sequence = self.get_sequences(cursor, f.m2m_db_table())
150 sequence_list.extend(
151 sequence or [{"table": f.m2m_db_table(), "column": None}]
152 )
153 return sequence_list
155 def get_sequences(self, cursor, table_name, table_fields=()):
156 """
157 Return a list of introspected sequences for table_name. Each sequence
158 is a dict: {'table': <table_name>, 'column': <column_name>}. An optional
159 'name' key can be added if the backend supports named sequences.
160 """
161 raise NotImplementedError(
162 "subclasses of BaseDatabaseIntrospection may require a get_sequences() "
163 "method"
164 )
166 def get_relations(self, cursor, table_name):
167 """
168 Return a dictionary of
169 {field_name: (field_name_other_table, other_table)} representing all
170 relationships to the given table.
171 """
172 raise NotImplementedError(
173 "subclasses of BaseDatabaseIntrospection may require a "
174 "get_relations() method."
175 )
177 def get_key_columns(self, cursor, table_name):
178 """
179 Backends can override this to return a list of:
180 (column_name, referenced_table_name, referenced_column_name)
181 for all key columns in given table.
182 """
183 raise NotImplementedError(
184 "subclasses of BaseDatabaseIntrospection may require a get_key_columns() "
185 "method"
186 )
188 def get_primary_key_column(self, cursor, table_name):
189 """
190 Return the name of the primary key column for the given table.
191 """
192 for constraint in self.get_constraints(cursor, table_name).values():
193 if constraint["primary_key"]:
194 return constraint["columns"][0]
195 return None
197 def get_constraints(self, cursor, table_name):
198 """
199 Retrieve any constraints or keys (unique, pk, fk, check, index)
200 across one or more columns.
202 Return a dict mapping constraint names to their attributes,
203 where attributes is a dict with keys:
204 * columns: List of columns this covers
205 * primary_key: True if primary key, False otherwise
206 * unique: True if this is a unique constraint, False otherwise
207 * foreign_key: (table, column) of target, or None
208 * check: True if check constraint, False otherwise
209 * index: True if index, False otherwise.
210 * orders: The order (ASC/DESC) defined for the columns of indexes
211 * type: The type of the index (btree, hash, etc.)
213 Some backends may return special constraint names that don't exist
214 if they don't name constraints of a certain type (e.g. SQLite)
215 """
216 raise NotImplementedError(
217 "subclasses of BaseDatabaseIntrospection may require a get_constraints() "
218 "method"
219 )