tests/dummy.py

   1 from datasette_connectors.row import Row
   2
   3
   4 _connector_type = 'dummy'
   5
   6 def inspect(path):
   7     tables = {}
   8     views = []
   9
  10     for table in ['table1', 'table2']:
  11         tables[table] = {
  12             'name': table,
  13             'columns': ['c1', 'c2', 'c3'],
  14             'primary_keys': [],
  15             'count': 2,
  16             'label_column': None,
  17             'hidden': False,
  18             'fts_table': None,
  19             'foreign_keys': {'incoming': [], 'outgoing': []},
  20         }
  21
  22     return tables, views, _connector_type
  23
  24
  25 class Connection:
  26     def __init__(self, path):
  27         self.path = path
  28
  29     def execute(self, sql, params=None, truncate=False, page_size=None, max_returned_rows=None):
  30         sql = sql.strip()
  31
  32         rows = []
  33         truncated = False
  34         description = []
  35
  36         if sql == 'select c1 from table1':
  37             rows = [
  38                 Row({'c1': 10}),
  39                 Row({'c1': 20})
  40             ]
  41             description = (('c1',),)
  42         elif sql == 'select rowid, * from table2 order by rowid limit 51':
  43             rows = [
  44                 Row({'rowid': 1, 'c1': 100, 'c2': 120, 'c3': 130}),
  45                 Row({'rowid': 2, 'c1': 200, 'c2': 220, 'c3': 230})
  46             ]
  47             description = (('rowid',), ('c1',), ('c2',), ('c3',))
  48         elif sql == 'select count(*) from table2':
  49             rows = [Row({'count(*)': 2})]
  50             description = (('count(*)',),)
  51         elif sql == """select distinct rowid from table2
  52                         where rowid is not null
  53                         limit 31""":
  54             rows = [
  55                 Row({'rowid': 1}),
  56                 Row({'rowid': 2})
  57             ]
  58             description = (('rowid',),)
  59         elif sql == """select distinct c1 from table2
  60                         where c1 is not null
  61                         limit 31""":
  62             rows = [
  63                 Row({'c1': 100}),
  64                 Row({'c1': 200})
  65             ]
  66             description = (('c1',),)
  67         elif sql == """select distinct c2 from table2
  68                         where c2 is not null
  69                         limit 31""":
  70             rows = [
  71                 Row({'c2': 120}),
  72                 Row({'c2': 220})
  73             ]
  74             description = (('c2',),)
  75         elif sql == """select distinct c3 from table2
  76                         where c3 is not null
  77                         limit 31""":
  78             rows = [
  79                 Row({'c3': 130}),
  80                 Row({'c3': 230})
  81             ]
  82             description = (('c3',),)
  83         elif sql == 'select sql from sqlite_master where name = :n and type=:t':
  84             if params['t'] != 'view':
  85                 rows = [Row({'sql': 'CREATE TABLE ' + params['n'] + ' (c1, c2, c3)'})]
  86                 description = (('sql',),)
  87         else:
  88             raise Exception("Unexpected query: %s" % sql)
  89
  90         return rows, truncated, description