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         'bigint': 'BigIntegerField',

    20         'integer unsigned': 'PositiveIntegerField',

    21         'decimal': 'DecimalField',

    22         'real': 'FloatField',

    23         'text': 'TextField',

    24         'char': 'CharField',

    25         'date': 'DateField',

    26         'datetime': 'DateTimeField',

    27         'time': 'TimeField',

    28     }

    29 

    30     def __getitem__(self, key):

    31         key = key.lower()

    32         try:

    33             return self.base_data_types_reverse[key]

    34         except KeyError:

    35             import re

    36             m = re.search(r'^\s*(?:var)?char\s*\(\s*(\d+)\s*\)\s*$', key)

    37             if m:

    38                 return ('CharField', {'max_length': int(m.group(1))})

    39             raise KeyError

    40 

    41 class DatabaseIntrospection(BaseDatabaseIntrospection):

    42     data_types_reverse = FlexibleFieldLookupDict()

    43 

    44     def get_table_list(self, cursor):

    45         "Returns a list of table names in the current database."

    46         # Skip the sqlite_sequence system table used for autoincrement key

    47         # generation.

    48         cursor.execute("""

    49             SELECT name FROM sqlite_master

    50             WHERE type='table' AND NOT name='sqlite_sequence'

    51             ORDER BY name""")

    52         return [row[0] for row in cursor.fetchall()]

    53 

    54     def get_table_description(self, cursor, table_name):

    55         "Returns a description of the table, with the DB-API cursor.description interface."

    56         return [(info['name'], info['type'], None, None, None, None,

    57                  info['null_ok']) for info in self._table_info(cursor, table_name)]

    58 

    59     def get_relations(self, cursor, table_name):

    60         """

    61         Returns a dictionary of {field_index: (field_index_other_table, other_table)}

    62         representing all relationships to the given table. Indexes are 0-based.

    63         """

    64 

    65         # Dictionary of relations to return

    66         relations = {}

    67 

    68         # Schema for this table

    69         cursor.execute("SELECT sql FROM sqlite_master WHERE tbl_name = %s AND type = %s", [table_name, "table"])

    70         results = cursor.fetchone()[0].strip()

    71         results = results[results.index('(')+1:results.rindex(')')]

    72 

    73         # Walk through and look for references to other tables. SQLite doesn't

    74         # really have enforced references, but since it echoes out the SQL used

    75         # to create the table we can look for REFERENCES statements used there.

    76         for field_index, field_desc in enumerate(results.split(',')):

    77             field_desc = field_desc.strip()

    78             if field_desc.startswith("UNIQUE"):

    79                 continue

    80 

    81             m = re.search('references (.*) \(["|](.*)["|]\)', field_desc, re.I)

    82             if not m:

    83                 continue

    84 

    85             table, column = [s.strip('"') for s in m.groups()]

    86 

    87             cursor.execute("SELECT sql FROM sqlite_master WHERE tbl_name = %s", [table])

    88             result = cursor.fetchone()

    89             if not result:

    90                 continue

    91             other_table_results = result[0].strip()

    92             li, ri = other_table_results.index('('), other_table_results.rindex(')')

    93             other_table_results = other_table_results[li+1:ri]

    94 

    95 

    96             for other_index, other_desc in enumerate(other_table_results.split(',')):

    97                 other_desc = other_desc.strip()

    98                 if other_desc.startswith('UNIQUE'):

    99                     continue

   100 

   101                 name = other_desc.split(' ', 1)[0].strip('"')

   102                 if name == column:

   103                     relations[field_index] = (other_index, table)

   104                     break

   105 

   106         return relations

   107 

   108     def get_indexes(self, cursor, table_name):

   109         """

   110         Returns a dictionary of fieldname -> infodict for the given table,

   111         where each infodict is in the format:

   112             {'primary_key': boolean representing whether it's the primary key,

   113              'unique': boolean representing whether it's a unique index}

   114         """

   115         indexes = {}

   116         for info in self._table_info(cursor, table_name):

   117             indexes[info['name']] = {'primary_key': info['pk'] != 0,

   118                                      'unique': False}

   119         cursor.execute('PRAGMA index_list(%s)' % self.connection.ops.quote_name(table_name))

   120         # seq, name, unique

   121         for index, unique in [(field[1], field[2]) for field in cursor.fetchall()]:

   122             if not unique:

   123                 continue

   124             cursor.execute('PRAGMA index_info(%s)' % self.connection.ops.quote_name(index))

   125             info = cursor.fetchall()

   126             # Skip indexes across multiple fields

   127             if len(info) != 1:

   128                 continue

   129             name = info[0][2] # seqno, cid, name

   130             indexes[name]['unique'] = True

   131         return indexes

   132 

   133     def _table_info(self, cursor, name):

   134         cursor.execute('PRAGMA table_info(%s)' % self.connection.ops.quote_name(name))

   135         # cid, name, type, notnull, dflt_value, pk

   136         return [{'name': field[1],

   137                  'type': field[2],

   138                  'null_ok': not field[3],

   139                  'pk': field[5]     # undocumented

   140                  } for field in cursor.fetchall()]

   141