tests/dummy.py

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