sqlite3tricks-py/test.py

#!/usr/bin/env python3

from __init__ import *

class RangeIterable(VTableIterable):
	def __init__(self, start=0, stop=float('inf'), step=1):
		self.start = start
		self.stop = stop
		self.step = step
		self.curr = self.start
	def __iter__(self):
		print(self.__class__.__name__, self)
		return self
	def __next__(self):
		if self.curr <= self.stop:
			res = self.curr
			self.curr += self.step
			return (res,)
		else:
			raise StopIteration

class RangeIterableGenerator(VTableIterable):
	def __init__(self, start=0, stop=float('inf'), step=1):
		self.start = start
		self.stop = stop
		self.step = step
	def __iter__(self):
		print(self.__class__.__name__, self)
		curr = self.start
		while curr <= self.stop:
			yield (curr,)
			curr += self.step

class RangeIterableMember(object):
	def __init__(self, extra):
		self.extra = extra
	@vtab_iterable(name='RangeIterableMember')
	def generator(self, start=0, stop=float('inf'), step=1):
		print(self.__class__.__name__, self, self.generator, self.extra)
		for curr in range(start, stop, step):
			yield (curr,)

class RangeIterableFancy(VTableIterable):
	_vtname_ = 'RangeIterableRenamed'
	_vtparams_ = ('start', 'stop', 'step', 'extra')
	_vtcolumns_ = ('once', 'twice', 'thrice')
	def __init__(self, start=0, stop=float('inf'), step=1, **kw):
		self.start = start
		self.stop = stop
		self.step = step
		self.kw = kw
	def __iter__(self):
		print(self.__class__.__name__, self, self.kw)
		curr = self.start
		while curr <= self.stop:
			yield (curr, curr*2, curr*3)
			curr += self.step

@vtab_iterable
def RangeGenerator(start=0, stop=float('inf'), step=1):
	print(RangeGenerator.__name__, RangeGenerator)
	for curr in range(start, stop, step):
		yield (curr,)

@vtab_iterable(name='RangeGeneratorRenamed', params=('start', 'stop', 'step', 'extra'), columns=('once', 'twice', 'thrice'))
def RangeGeneratorFancy(start=0, stop=float('inf'), step=1, **kw):
	print(RangeGeneratorFancy.__name__, RangeGeneratorFancy, kw)
	for curr in range(start, stop, step):
		yield (curr, curr*2, curr*3)

if __name__ == '__main__':
	import sqlite3

	conn = sqlite3.connect(':memory:')
	pDB = connect(conn)

	conn.execute('CREATE TABLE test (a INTEGER, b INTEGER)')
	conn.execute('INSERT INTO test (a, b) VALUES (?, ?)', (1, 2))
	conn.execute('INSERT INTO test (a, b) VALUES (?, ?)', (3, 4))
	conn.execute('INSERT INTO test (a, b) VALUES (?, ?)', (5, 6))
	conn.execute('INSERT INTO test (a, b) VALUES (?, ?)', (7, 8))

	for row in conn.execute('SELECT * FROM test'):
		print('sqlite3:', row)
	for row in pDB.execute('SELECT * FROM test'):
		print('tricks:', row)
	print()

	for row in conn.execute('SELECT * FROM test WHERE a = ?', (1,)):
		print('sqlite3:', row)
	for row in pDB.execute('SELECT * FROM test WHERE a = ?', (1,)):
		print('tricks:', row)
	print()

	RangeIterable.register(conn)
	for row in conn.execute('SELECT * FROM RangeIterable(10, 16, 2)'):
		print('sqlite3:', row)
	for row in pDB.execute('SELECT * FROM RangeIterable(10, 16, 2)'):
		print('tricks:', row)
	print()

	RangeIterableGenerator.register(conn)
	for row in conn.execute('SELECT * FROM RangeIterableGenerator(20, 26, 2)'):
		print('sqlite3:', row)
	for row in pDB.execute('SELECT * FROM RangeIterableGenerator(20, 26, 2)'):
		print('tricks:', row)
	print()

	obj = RangeIterableMember('hello')
	obj.generator.register(conn)
	for row in conn.execute('SELECT * FROM RangeIterableMember(30, 36, 2)'):
		print('sqlite3:', row)
	for row in pDB.execute('SELECT * FROM RangeIterableMember(30, 36, 2)'):
		print('tricks:', row)
	print()

	RangeIterableFancy.register(conn)
	for row in conn.execute('SELECT * FROM RangeIterableRenamed(40, 46, 2, datetime())'):
		print('sqlite3:', row)
	for row in pDB.execute('SELECT * FROM RangeIterableRenamed(40, 46, 2, datetime())'):
		print('tricks:', row)
	print()

	RangeGenerator.register(conn)
	for row in conn.execute('SELECT * FROM RangeGenerator(50, 56, 2)'):
		print('sqlite3:', row)
	for row in pDB.execute('SELECT * FROM RangeGenerator(50, 56, 2)'):
		print('tricks:', row)
	print()

	RangeGeneratorFancy.register(conn)
	for row in conn.execute('SELECT * FROM RangeGeneratorRenamed(60, 66, 2, datetime())'):
		print('sqlite3:', row)
	for row in pDB.execute('SELECT * FROM RangeGeneratorRenamed(60, 66, 2, datetime())'):
		print('tricks:', row)
	print()

	import hashlib

	async def md5sum(t):
		return hashlib.md5(str(t).encode('utf-8')).hexdigest()

	create_function(conn, 'md5', 1, md5sum)

	for row in conn.execute('SELECT md5(value) FROM RangeIterable(10, 16, 2)'):
		print(row)
	for row in pDB.execute('SELECT md5(value) AS md5_value FROM RangeIterable(10, 16, 2)'):
		print(row)