|
1 import re |
|
2 from django.db.backends import BaseDatabaseIntrospection |
|
3 |
|
4 # This light wrapper "fakes" a dictionary interface, because some SQLite data |
|
5 # types include variables in them -- e.g. "varchar(30)" -- and can't be matched |
|
6 # as a simple dictionary lookup. |
|
7 class FlexibleFieldLookupDict: |
|
8 # Maps SQL types to Django Field types. Some of the SQL types have multiple |
|
9 # entries here because SQLite allows for anything and doesn't normalize the |
|
10 # field type; it uses whatever was given. |
|
11 base_data_types_reverse = { |
|
12 'bool': 'BooleanField', |
|
13 'boolean': 'BooleanField', |
|
14 'smallint': 'SmallIntegerField', |
|
15 'smallint unsigned': 'PositiveSmallIntegerField', |
|
16 'smallinteger': 'SmallIntegerField', |
|
17 'int': 'IntegerField', |
|
18 'integer': 'IntegerField', |
|
19 'integer unsigned': 'PositiveIntegerField', |
|
20 'decimal': 'DecimalField', |
|
21 'real': 'FloatField', |
|
22 'text': 'TextField', |
|
23 'char': 'CharField', |
|
24 'date': 'DateField', |
|
25 'datetime': 'DateTimeField', |
|
26 'time': 'TimeField', |
|
27 } |
|
28 |
|
29 def __getitem__(self, key): |
|
30 key = key.lower() |
|
31 try: |
|
32 return self.base_data_types_reverse[key] |
|
33 except KeyError: |
|
34 import re |
|
35 m = re.search(r'^\s*(?:var)?char\s*\(\s*(\d+)\s*\)\s*$', key) |
|
36 if m: |
|
37 return ('CharField', {'max_length': int(m.group(1))}) |
|
38 raise KeyError |
|
39 |
|
40 class DatabaseIntrospection(BaseDatabaseIntrospection): |
|
41 data_types_reverse = FlexibleFieldLookupDict() |
|
42 |
|
43 def get_table_list(self, cursor): |
|
44 "Returns a list of table names in the current database." |
|
45 # Skip the sqlite_sequence system table used for autoincrement key |
|
46 # generation. |
|
47 cursor.execute(""" |
|
48 SELECT name FROM sqlite_master |
|
49 WHERE type='table' AND NOT name='sqlite_sequence' |
|
50 ORDER BY name""") |
|
51 return [row[0] for row in cursor.fetchall()] |
|
52 |
|
53 def get_table_description(self, cursor, table_name): |
|
54 "Returns a description of the table, with the DB-API cursor.description interface." |
|
55 return [(info['name'], info['type'], None, None, None, None, |
|
56 info['null_ok']) for info in self._table_info(cursor, table_name)] |
|
57 |
|
58 def get_relations(self, cursor, table_name): |
|
59 """ |
|
60 Returns a dictionary of {field_index: (field_index_other_table, other_table)} |
|
61 representing all relationships to the given table. Indexes are 0-based. |
|
62 """ |
|
63 |
|
64 # Dictionary of relations to return |
|
65 relations = {} |
|
66 |
|
67 # Schema for this table |
|
68 cursor.execute("SELECT sql FROM sqlite_master WHERE tbl_name = %s", [table_name]) |
|
69 results = cursor.fetchone()[0].strip() |
|
70 results = results[results.index('(')+1:results.rindex(')')] |
|
71 |
|
72 # Walk through and look for references to other tables. SQLite doesn't |
|
73 # really have enforced references, but since it echoes out the SQL used |
|
74 # to create the table we can look for REFERENCES statements used there. |
|
75 for field_index, field_desc in enumerate(results.split(',')): |
|
76 field_desc = field_desc.strip() |
|
77 if field_desc.startswith("UNIQUE"): |
|
78 continue |
|
79 |
|
80 m = re.search('references (.*) \(["|](.*)["|]\)', field_desc, re.I) |
|
81 if not m: |
|
82 continue |
|
83 |
|
84 table, column = [s.strip('"') for s in m.groups()] |
|
85 |
|
86 cursor.execute("SELECT sql FROM sqlite_master WHERE tbl_name = %s", [table]) |
|
87 result = cursor.fetchone() |
|
88 if not result: |
|
89 continue |
|
90 other_table_results = result[0].strip() |
|
91 li, ri = other_table_results.index('('), other_table_results.rindex(')') |
|
92 other_table_results = other_table_results[li+1:ri] |
|
93 |
|
94 |
|
95 for other_index, other_desc in enumerate(other_table_results.split(',')): |
|
96 other_desc = other_desc.strip() |
|
97 if other_desc.startswith('UNIQUE'): |
|
98 continue |
|
99 |
|
100 name = other_desc.split(' ', 1)[0].strip('"') |
|
101 if name == column: |
|
102 relations[field_index] = (other_index, table) |
|
103 break |
|
104 |
|
105 return relations |
|
106 |
|
107 def get_indexes(self, cursor, table_name): |
|
108 """ |
|
109 Returns a dictionary of fieldname -> infodict for the given table, |
|
110 where each infodict is in the format: |
|
111 {'primary_key': boolean representing whether it's the primary key, |
|
112 'unique': boolean representing whether it's a unique index} |
|
113 """ |
|
114 indexes = {} |
|
115 for info in self._table_info(cursor, table_name): |
|
116 indexes[info['name']] = {'primary_key': info['pk'] != 0, |
|
117 'unique': False} |
|
118 cursor.execute('PRAGMA index_list(%s)' % self.connection.ops.quote_name(table_name)) |
|
119 # seq, name, unique |
|
120 for index, unique in [(field[1], field[2]) for field in cursor.fetchall()]: |
|
121 if not unique: |
|
122 continue |
|
123 cursor.execute('PRAGMA index_info(%s)' % self.connection.ops.quote_name(index)) |
|
124 info = cursor.fetchall() |
|
125 # Skip indexes across multiple fields |
|
126 if len(info) != 1: |
|
127 continue |
|
128 name = info[0][2] # seqno, cid, name |
|
129 indexes[name]['unique'] = True |
|
130 return indexes |
|
131 |
|
132 def _table_info(self, cursor, name): |
|
133 cursor.execute('PRAGMA table_info(%s)' % self.connection.ops.quote_name(name)) |
|
134 # cid, name, type, notnull, dflt_value, pk |
|
135 return [{'name': field[1], |
|
136 'type': field[2], |
|
137 'null_ok': not field[3], |
|
138 'pk': field[5] # undocumented |
|
139 } for field in cursor.fetchall()] |
|
140 |