1from collections import namedtuple
2
3# Structure returned by DatabaseIntrospection.get_table_list()
4TableInfo = namedtuple("TableInfo", ["name", "type"])
5
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)
12
13
14class BaseDatabaseIntrospection:
15 """Encapsulate backend-specific introspection utilities."""
16
17 data_types_reverse = {}
18
19 def __init__(self, connection):
20 self.connection = connection
21
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 Plain field type to a database column.
26
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]
31
32 def identifier_converter(self, name):
33 """
34 Apply a conversion to the identifier for the purposes of comparison.
35
36 The default identifier converter is for case sensitive comparison.
37 """
38 return name
39
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 """
47
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 )
54
55 if cursor is None:
56 with self.connection.cursor() as cursor:
57 return get_names(cursor)
58 return get_names(cursor)
59
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 )
69
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 )
79
80 def get_migratable_models(self):
81 from plain.models import models_registry
82 from plain.packages import packages_registry
83
84 return (
85 model
86 for package_config in packages_registry.get_package_configs()
87 for model in models_registry.get_models(
88 package_label=package_config.package_label
89 )
90 if model._meta.can_migrate(self.connection)
91 )
92
93 def plain_table_names(self, only_existing=False, include_views=True):
94 """
95 Return a list of all table names that have associated Plain models and
96 are in INSTALLED_PACKAGES.
97
98 If only_existing is True, include only the tables in the database.
99 """
100 tables = set()
101 for model in self.get_migratable_models():
102 tables.add(model._meta.db_table)
103 tables.update(f.m2m_db_table() for f in model._meta.local_many_to_many)
104 tables = list(tables)
105 if only_existing:
106 existing_tables = set(self.table_names(include_views=include_views))
107 tables = [
108 t for t in tables if self.identifier_converter(t) in existing_tables
109 ]
110 return tables
111
112 def sequence_list(self):
113 """
114 Return a list of information about all DB sequences for all models in
115 all packages.
116 """
117 sequence_list = []
118 with self.connection.cursor() as cursor:
119 for model in self.get_migratable_models():
120 sequence_list.extend(
121 self.get_sequences(
122 cursor, model._meta.db_table, model._meta.local_fields
123 )
124 )
125 return sequence_list
126
127 def get_sequences(self, cursor, table_name, table_fields=()):
128 """
129 Return a list of introspected sequences for table_name. Each sequence
130 is a dict: {'table': <table_name>, 'column': <column_name>}. An optional
131 'name' key can be added if the backend supports named sequences.
132 """
133 raise NotImplementedError(
134 "subclasses of BaseDatabaseIntrospection may require a get_sequences() "
135 "method"
136 )
137
138 def get_relations(self, cursor, table_name):
139 """
140 Return a dictionary of {field_name: (field_name_other_table, other_table)}
141 representing all foreign keys in the given table.
142 """
143 raise NotImplementedError(
144 "subclasses of BaseDatabaseIntrospection may require a "
145 "get_relations() method."
146 )
147
148 def get_primary_key_column(self, cursor, table_name):
149 """
150 Return the name of the primary key column for the given table.
151 """
152 columns = self.get_primary_key_columns(cursor, table_name)
153 return columns[0] if columns else None
154
155 def get_primary_key_columns(self, cursor, table_name):
156 """Return a list of primary key columns for the given table."""
157 for constraint in self.get_constraints(cursor, table_name).values():
158 if constraint["primary_key"]:
159 return constraint["columns"]
160 return None
161
162 def get_constraints(self, cursor, table_name):
163 """
164 Retrieve any constraints or keys (unique, pk, fk, check, index)
165 across one or more columns.
166
167 Return a dict mapping constraint names to their attributes,
168 where attributes is a dict with keys:
169 * columns: List of columns this covers
170 * primary_key: True if primary key, False otherwise
171 * unique: True if this is a unique constraint, False otherwise
172 * foreign_key: (table, column) of target, or None
173 * check: True if check constraint, False otherwise
174 * index: True if index, False otherwise.
175 * orders: The order (ASC/DESC) defined for the columns of indexes
176 * type: The type of the index (btree, hash, etc.)
177
178 Some backends may return special constraint names that don't exist
179 if they don't name constraints of a certain type (e.g. SQLite)
180 """
181 raise NotImplementedError(
182 "subclasses of BaseDatabaseIntrospection may require a get_constraints() "
183 "method"
184 )