<gh_stars>0
import sqlite3
import sqlalchemy
import enum
import collections
import recordclass
import traceback
import itertools
import signal
import sys

from pathlib import Path
from tabulate import tabulate
from tqdm import tqdm


class ProcessorState(enum.Enum):
	queued = 1
	running = 2
	failed = 3
	done = 4
	ignored = 5


Slot = collections.namedtuple(
	'Slot', ['name', 'sql_name', 'sql_type'])


class Task:
	def __init__(self, input_, processors=None):
		self._input = str(input_)

		if processors is None or processors == "all":
			self._processors = "all"
		else:
			self._processors = tuple(processors)

	@property
	def input_(self):
		return self._input

	@property
	def processors(self):
		return self._processors


def _chunks(l, n):
	for i in range(0, len(l), n):
		yield l[i:i + n]


def _check_disjunct_outputs(processors):
	outputs = set()
	for processor in processors:
		added = set(processor.outputs)
		shared = added & outputs
		if len(shared) > 0:
			raise ValueError(
				"processor output(s) %s must not be generated by multiple processors" % str(shared))
		outputs |= added


SLOT_COL = 's_%s'
PROCESS_STATE_COL = 'p_st_%s'
PROCESS_ITERATION_COL = 'p_it_%s'


class InterruptedException(Exception):
	pass

class NeedsMigration(Exception):
	pass


class Interruptions:
	def __init__(self):
		self.allowed = False
		self.requested = False


class Migration:
	def __init__(self, sis, engine):
		self._sis = sis
		self._engine = engine

		self._engine.execute("DROP TABLE IF EXISTS migrated_task")

		metadata = sqlalchemy.MetaData(engine)
		metadata.reflect(bind=engine)
		reflected_task_table = sqlalchemy.Table(
			'task', metadata, autoload=True, autoload_with=engine)

		existing = set([c.name for c in reflected_task_table.columns])
		needed = set([c.name for c in sis._task_table.columns])

		self._current_task_cols = existing
		self._needed_task_cols = needed

		self._reflected_task_table = reflected_task_table
		self._metadata = metadata

	@property
	def is_empty(self):
		if len(self._needed_task_cols - self._current_task_cols) > 0:
			return False
		return True

	def run(self):
		conn = self._engine.connect()

		tt0 = self._sis._task_table

		tt1 = self._sis._create_task_table("migrated_task", self._metadata)
		self._metadata.create_all()

		try:
			trans = conn.begin()
			try:
				cols = list(self._needed_task_cols & self._current_task_cols)

				print("migrating columns %s." % str(cols))

				sel = sqlalchemy.select([getattr(tt0.c, s) for s in cols])
				stmt = tt1.insert().from_select(cols, sel)
				conn.execute(stmt)

				trans.commit()

			except:
				trans.rollback()
				raise

			self._engine.execute("DROP TABLE task")
			self._engine.execute("ALTER TABLE migrated_task RENAME TO task")

		finally:
			conn.close()

		print("done.")
		sys.exit(0)


class Sisyphos:
	def __init__(self, path, slots, processors):
		self._slots = dict((a.name, a) for a in slots)
		self._processors = dict((p.name, p) for p in processors)

		if len(set([a.sql_name for a in slots])) != len(slots):
			raise ValueError("slots' sql_names need to be unique: %s" % str([a.sql_name for a in slots]))

		if len(set([p.sql_name for p in processors])) != len(processors):
			raise ValueError("processors' sql_names need to be unique: %s" % str([p.sql_name for p in processors]))

		_check_disjunct_outputs(processors)

		for processor in processors:
			for x in processor.outputs:
				if x not in self._slots:
					raise ValueError("unknown slot '%s' referenced in output of processor '%s'" % (
						x, processor.name))
			for x in processor.inputs:
				if x not in self._slots:
					raise ValueError("unknown slot '%s' referenced in input of processor '%s'" % (
						x, processor.name))

		self._slot_sql_name = dict((a.name, a.sql_name) for a in slots)

		db_uri = 'sqlite:///%s' % str(Path(path))
		self._engine = sqlalchemy.create_engine(db_uri, isolation_level="SERIALIZABLE")

		# see https://docs.sqlalchemy.org/en/13/dialects/sqlite.html#pysqlite-serializable
		@sqlalchemy.event.listens_for(self._engine, "connect")
		def do_connect(dbapi_connection, connection_record):
		    # disable pysqlite's emitting of the BEGIN statement entirely.
		    # also stops it from emitting COMMIT before any DDL.
		    dbapi_connection.isolation_level = None

		@sqlalchemy.event.listens_for(self._engine, "begin")
		def do_begin(conn):
		    conn.execute("BEGIN EXCLUSIVE")

		metadata = sqlalchemy.MetaData(self._engine)

		self._task_table = self._create_task_table("task", metadata)

		self._data_table = dict()
		for a in slots:
			self._data_table[a.sql_name] = sqlalchemy.Table(SLOT_COL % a.sql_name, metadata,
				sqlalchemy.Column(
					'id', sqlalchemy.Integer, primary_key=True),
				sqlalchemy.Column(
					'data', sqlalchemy.LargeBinary if a.sql_type is None else a.sql_type))

		self._err_table = dict()
		for p in processors:
			self._err_table[p.sql_name] = sqlalchemy.Table('err_%s' % p.sql_name, metadata,
				sqlalchemy.Column(
					'id', sqlalchemy.Integer, primary_key=True),
				sqlalchemy.Column(
					'err', sqlalchemy.Text),
				extend_existing=True)

		metadata.create_all()

		self._migration = self._build_migration()

	def _create_task_table(self, name, metadata):
		tt = sqlalchemy.Table(name, metadata,
			sqlalchemy.Column(
				'id', sqlalchemy.Integer,
				sqlalchemy.Sequence('task_id_seq', metadata=metadata), primary_key=True),
			sqlalchemy.Column(
				'input', sqlalchemy.String,
				nullable=False, index=True, unique=True),
			*self._task_table_columns()
		)

		for p in self._processors.values():
			columns = [getattr(tt.c, PROCESS_STATE_COL % p.sql_name)] + [
				getattr(tt.c, SLOT_COL % self._slot_sql_name[x]) for x in p.inputs] + [
				getattr(tt.c, PROCESS_ITERATION_COL % p.sql_name)]
			sqlalchemy.Index('idx_proc_%s' % p, *columns)

		return tt

	def _task_table_columns(self):
		return list(itertools.chain(
			[sqlalchemy.Column(
				SLOT_COL % a.sql_name, 
				sqlalchemy.Boolean,
				nullable=False, default=False) for a in self._slots.values()],
			[sqlalchemy.Column(
				PROCESS_STATE_COL % proc.sql_name, 
				sqlalchemy.Enum(ProcessorState),
				nullable=False, default=ProcessorState.queued) for proc in self._processors.values()],
			[sqlalchemy.Column(
				PROCESS_ITERATION_COL % proc.sql_name, 
				sqlalchemy.Integer,
				nullable=False, default=0) for proc in self._processors.values()]))

	def _build_migration(self):
		migration = Migration(self, self._engine)
		if migration.is_empty:
			migration = None
		return migration

	def needs_migration(self):
		return self._migration is not None

	def _ensure_is_migrated(self):
		if self.needs_migration():
			raise NeedsMigration()

	def migrate(self):
		if self._migration:
			self._migration.run()
		sys.exit(0)

	@property
	def processors(self):
		return [p.name for p in self._processors.values()]

	def _slot_column_name(self, public_slot_name):
		return SLOT_COL % self._slot_sql_name[public_slot_name]

	def add_tasks(self, tasks, chunk_size=100, write_size=100, show_progress=False):
		self._ensure_is_migrated()

		tt = self._task_table
		all_processor_sql_names = set([p.sql_name for p in self._processors.values()])

		conn = self._engine.connect()

		try:

			with tqdm(total=len(tasks), desc='adding tasks', disable=not show_progress) as pbar:
				data = []

				for chunked_tasks in _chunks(tasks, chunk_size):

					by_input = dict((c.input_, c) for c in chunked_tasks)

					selection = sqlalchemy.sql.select([tt.c.input]).where(
						tt.c.input.in_([c.input_ for c in chunked_tasks]))
					existing_inputs = set(
						[row['input'] for row in conn.execute(selection).fetchall()])

					for task_input in set(by_input.keys()) - existing_inputs:
						task_data = dict(input=task_input)

						task = by_input[task_input]
						processor_names = task.processors
						if processor_names != "all":
							assert type(processor_names) is tuple

							ignored_processors = all_processor_sql_names - set([
								self._processors[p].sql_name for p in processor_names])
						else:
							ignored_processors = set()

						for p in all_processor_sql_names:
							state = ProcessorState.ignored if p in ignored_processors else ProcessorState.queued
							task_data[PROCESS_STATE_COL % p] = state

						data.append(task_data)

					if len(data) >= write_size:
						conn.execute(self._task_table.insert(), data)
						data = []

					pbar.update(len(chunked_tasks))

				if data:
					conn.execute(self._task_table.insert(), data)

		finally:
			conn.close()

	def _run_processor_core(self, conn, processor, task_id, task_input, task_stage, input_slots, interruptions):
		tt = self._task_table

		args = input_slots.copy()
		args['task_input'] = task_input

		inputs = collections.namedtuple('Inputs', ['task_input', *input_slots.keys()])(**args)

		outputs = recordclass.recordclass('Outputs', processor.outputs)(*[None for _ in processor.outputs])
		
		try:
			interruptions.allowed = True
			processor.run(inputs, outputs)
		finally:
			interruptions.allowed = False

		if not all(getattr(outputs, x) is not None for x in processor.outputs):
			raise ValueError("processor %s did not produce all required output slots" % processor.name)

		trans = conn.begin()
		try:
			for slot_name in processor.outputs:
				data_table = self._data_table[self._slot_sql_name[slot_name]]
				conn.execute(data_table.insert().values(
					id=task_id,
					data=getattr(outputs, slot_name)))

			new_state = dict()
			new_state[PROCESS_ITERATION_COL % processor.sql_name] = task_stage + 1

			if task_stage + 1 >= getattr(processor, 'n_stages', 1):
				new_state[PROCESS_STATE_COL % processor.sql_name] = ProcessorState.done

			for output in processor.outputs:
				new_state[self._slot_column_name(output)] = True

			stmt = tt.update().values(**new_state).where(tt.c.input == task_input)
			conn.execute(stmt)

			trans.commit()

		except:
			trans.rollback()
			raise

	def _load_slots(self, conn, slot_names, task_id):
		data = dict()
		for slot_name in slot_names:
			data_table = self._data_table[self._slot_sql_name[slot_name]]
			selection = sqlalchemy.sql.select([data_table.c.data]).where(
				data_table.c.id == task_id)
			data[slot_name] = conn.execute(selection).fetchone()['data']
		return data

	def _run_processor(self, conn, processor, show_progress, interruptions):
		tt = self._task_table
		any_work_done = False

		condition = sqlalchemy.and_(*[
			getattr(tt.c, self._slot_column_name(c)) == True for c in processor.inputs] + [
			getattr(tt.c, PROCESS_STATE_COL % processor.sql_name) == ProcessorState.queued])

		def query_remaining():
			selection = sqlalchemy.sql.select(
				[sqlalchemy.func.count()]).select_from(tt).where(condition)
			return conn.execute(selection).scalar()

		total = query_remaining()
		rem = total

		if total == 0:
			return False

		with tqdm(desc="running %s" % processor.name, disable=not show_progress, total=total) as pbar:

			while not interruptions.requested:

				trans = conn.begin()
				try:
					selection = sqlalchemy.sql.select([tt], for_update=True).where(condition).order_by(
						getattr(tt.c, PROCESS_ITERATION_COL % processor.sql_name))
					row = conn.execute(selection).fetchone()

					if row is None:  # done with this processor?
						trans.rollback()
						break

					any_work_done = True

					task_input = row[tt.c.input]
					task_id = row[tt.c.id]
					task_stage = row[getattr(tt.c, PROCESS_ITERATION_COL % processor.sql_name)]

					input_slots_data = self._load_slots(conn, processor.inputs, task_id)

					new_state = dict()
					new_state[PROCESS_STATE_COL % processor.sql_name] = ProcessorState.running

					stmt = tt.update().values(**new_state).where(tt.c.input == task_input)
					conn.execute(stmt)

					trans.commit()

				except:
					trans.rollback()
					raise

				try:
					self._run_processor_core(
						conn, processor, task_id, task_input, task_stage, input_slots_data, interruptions)

				except InterruptedException as e:

					trans = conn.begin()
					try:
						new_state = dict()
						new_state[PROCESS_STATE_COL % processor.sql_name] = ProcessorState.queued

						stmt = tt.update().values(**new_state).where(tt.c.input == task_input)
						conn.execute(stmt)

						trans.commit()

					except:
						trans.rollback()
						raise

					raise

				except:
					trans = conn.begin()
					try:
						err_table = self._err_table[processor.sql_name]

						stmt = err_table.delete().where(
							err_table.c.id == task_id)
						conn.execute(stmt)

						stmt = err_table.insert().values(
							id=task_id, err=traceback.format_exc())
						conn.execute(stmt)

						new_state = dict()
						new_state[PROCESS_STATE_COL % processor.sql_name] = ProcessorState.failed

						stmt = tt.update().values(**new_state).where(tt.c.input == task_input)
						conn.execute(stmt)

						trans.commit()

					except:
						trans.rollback()
						raise

					raise

				if show_progress:
					new_rem = query_remaining()
					pbar.update(rem - new_rem)
					rem = new_rem

		if interruptions.requested:
			raise InterruptedException()

		return any_work_done

	def _resolve_processors(self, processor_names):
		if processor_names == 'all':
			processor_names = list(self._processors.keys())

		return [self._processors[name] for name in processor_names]

	def run(self, processor_names='all', show_progress=False):
		self._ensure_is_migrated()
		processors = self._resolve_processors(processor_names)

		interruptions = Interruptions()
		def handler(signum, frame):
			interruptions.requested = True
			if interruptions.allowed:
				raise InterruptedException()

		old_handler = signal.signal(signal.SIGINT, handler)

		try:
			conn = self._engine.connect()

			try:
				while True:
					any_work_done = False

					for processor in processors:
						if self._run_processor(
							conn, processor, show_progress, interruptions):
							
							any_work_done = True

					if not any_work_done:
						break

			finally:
				conn.close()

		finally:
			signal.signal(signal.SIGINT, old_handler)

	def reset(self, processor_names='all', reset_hard=False):
		self._ensure_is_migrated()
		tt = self._task_table

		processors = self._resolve_processors(processor_names)

		if not reset_hard:
			selected_states = [ProcessorState.failed]
		else:
			selected_states = [ProcessorState.failed, ProcessorState.done, ProcessorState.ignored]

		conn = self._engine.connect()

		try:
			for processor in processors:

				condition = getattr(tt.c, PROCESS_STATE_COL % processor.sql_name).in_(selected_states)

				selection = sqlalchemy.sql.select([tt], for_update=True).where(condition)

				trans = conn.begin()
				try:
					for task in conn.execute(selection).fetchall():
						task_id = task['id']

						err_table = self._err_table[processor.sql_name]
						stmt = sqlalchemy.sql.delete(err_table).where(
							err_table.c.id == task_id)
						conn.execute(stmt)

						for x in processor.outputs:
							data_table = self._data_table[self._slot_sql_name[x]]
							stmt = sqlalchemy.sql.delete(data_table).where(
								data_table.c.id == task_id)
							conn.execute(stmt)

						new_state = dict()
						new_state[PROCESS_STATE_COL % processor.sql_name] = ProcessorState.queued

						stmt = tt.update().values(**new_state).where(tt.c.id == task_id)
						conn.execute(stmt)

					trans.commit()

				except:
					trans.rollback()
					raise

		finally:
			conn.close()		

	def print_status(self):
		self._ensure_is_migrated()

		tt = self._task_table
		status_table = []

		conn = self._engine.connect()

		try:
			for processor in sorted(self._processors.values(), key=lambda p: len(p.inputs)):
				counts = dict()

				state_c = getattr(tt.c, PROCESS_STATE_COL % processor.sql_name)
				selection = sqlalchemy.sql.select(
					[state_c, sqlalchemy.func.count()]).select_from(tt).group_by(state_c)
				for status in conn.execute(selection).fetchall():
					state, count = list(status.values())
					counts[state] = count

				status_table.append(
					[processor.name] + [counts.get(s, 0) for s in ProcessorState])

			print(tabulate(status_table, headers=["processor"] + [s.name for s in ProcessorState], tablefmt="psql"))

		finally:
			conn.close()

	def tasks(self):
		self._ensure_is_migrated()

		tt = self._task_table
		conn = self._engine.connect()

		try:
			selection = sqlalchemy.sql.select([tt])
			for task in conn.execute(selection).fetchall():

				processors = dict()
				for p in self._processors.values():
					processors[p.name] = task[PROCESS_STATE_COL % p.sql_name]

				slots = dict()
				for s in self._slots.values():
					slots[s.name] = task[SLOT_COL % s.sql_name]

				yield task['input'], processors, slots

		finally:
			conn.close()

	def print_tasks(self, up_to=10):
		self._ensure_is_migrated()

		processors = sorted(p.name for p in self._processors.values())
		slots = sorted(p.name for p in self._slots.values())

		tasks_table = []
		for task_input, task_processors, task_slots in itertools.islice(self.tasks(), 0, up_to):
			tasks_table.append(list(itertools.chain(
				[task_input],
				[task_processors[x].name for x in processors],
				[task_slots[x] for x in slots])))

		print(tabulate(tasks_table, headers=["input"] + processors + slots, tablefmt="psql"))

	def outputs(self, processor_name):
		self._ensure_is_migrated()
		processor = self._processors[processor_name]

		tt = self._task_table
		conn = self._engine.connect()

		try:
			condition = getattr(tt.c, PROCESS_STATE_COL % processor.sql_name) == ProcessorState.done

			selection = sqlalchemy.sql.select([tt]).where(condition)
			for task in conn.execute(selection).fetchall():
				yield task['input'], self._load_slots(conn, processor.outputs, task['id'])

		finally:
			conn.close()

	def print_outputs(self, processor_name, up_to=10):
		self._ensure_is_migrated()
		processor = self._processors[processor_name]
		outputs = processor.outputs

		outputs_table = []
		for task_input, output_data in itertools.islice(self.outputs(processor_name), 0, up_to):
			outputs_table.append([task_input] + [output_data[x] for x in outputs])

		print(tabulate(outputs_table, headers=["input"] + outputs, tablefmt="psql"))

	def slot(self, slot_name, task_input):
		self._ensure_is_migrated()
		slot = self._slots[slot_name]
		tt = self._task_table

		conn = self._engine.connect()

		try:
			selection = sqlalchemy.sql.select([tt]).where(tt.c.input == task_input)
			task = conn.execute(selection).fetchone()
			if task is None:
				raise ValueError("task '%s' not found" % task_input)

			dt = self._data_table[slot.sql_name]
			selection = sqlalchemy.sql.select([dt]).where(dt.c.id == task["id"])
			data = conn.execute(selection).fetchone()
			if data is None:
				raise ValueError("no data in slot '%s' for task '%s'" % (slot_name, task_input))

			return data["data"]

		finally:
			conn.close()


class Processor:
	def __init__(self, name, sql_name):
		self._name = name
		self._sql_name = sql_name

	def run(self, task, **kwargs):
		raise NotImplementedError()

	@property
	def name(self):
		return self._name

	@property
	def sql_name(self):
		return self._sql_name

import sqlite3
import sqlalchemy
import enum
import collections
import recordclass
import traceback
import itertools
import signal
import sys

from pathlib import Path
from tabulate import tabulate
from tqdm import tqdm


class ProcessorState(enum.Enum):
	queued = 1
	running = 2
	failed = 3
	done = 4
	ignored = 5


Slot = collections.namedtuple(
	'Slot', ['name', 'sql_name', 'sql_type'])


class Task:
	def __init__(self, input_, processors=None):
		self._input = str(input_)

		if processors is None or processors == "all":
			self._processors = "all"
		else:
			self._processors = tuple(processors)

	@property
	def input_(self):
		return self._input

	@property
	def processors(self):
		return self._processors


def _chunks(l, n):
	for i in range(0, len(l), n):
		yield l[i:i + n]


def _check_disjunct_outputs(processors):
	outputs = set()
	for processor in processors:
		added = set(processor.outputs)
		shared = added & outputs
		if len(shared) > 0:
			raise ValueError(
				"processor output(s) %s must not be generated by multiple processors" % str(shared))
		outputs |= added


SLOT_COL = 's_%s'
PROCESS_STATE_COL = 'p_st_%s'
PROCESS_ITERATION_COL = 'p_it_%s'


class InterruptedException(Exception):
	pass

class NeedsMigration(Exception):
	pass


class Interruptions:
	def __init__(self):
		self.allowed = False
		self.requested = False


class Migration:
	def __init__(self, sis, engine):
		self._sis = sis
		self._engine = engine

		self._engine.execute("DROP TABLE IF EXISTS migrated_task")

		metadata = sqlalchemy.MetaData(engine)
		metadata.reflect(bind=engine)
		reflected_task_table = sqlalchemy.Table(
			'task', metadata, autoload=True, autoload_with=engine)

		existing = set([c.name for c in reflected_task_table.columns])
		needed = set([c.name for c in sis._task_table.columns])

		self._current_task_cols = existing
		self._needed_task_cols = needed

		self._reflected_task_table = reflected_task_table
		self._metadata = metadata

	@property
	def is_empty(self):
		if len(self._needed_task_cols - self._current_task_cols) > 0:
			return False
		return True

	def run(self):
		conn = self._engine.connect()

		tt0 = self._sis._task_table

		tt1 = self._sis._create_task_table("migrated_task", self._metadata)
		self._metadata.create_all()

		try:
			trans = conn.begin()
			try:
				cols = list(self._needed_task_cols & self._current_task_cols)

				print("migrating columns %s." % str(cols))

				sel = sqlalchemy.select([getattr(tt0.c, s) for s in cols])
				stmt = tt1.insert().from_select(cols, sel)
				conn.execute(stmt)

				trans.commit()

			except:
				trans.rollback()
				raise

			self._engine.execute("DROP TABLE task")
			self._engine.execute("ALTER TABLE migrated_task RENAME TO task")

		finally:
			conn.close()

		print("done.")
		sys.exit(0)


class Sisyphos:
	def __init__(self, path, slots, processors):
		self._slots = dict((a.name, a) for a in slots)
		self._processors = dict((p.name, p) for p in processors)

		if len(set([a.sql_name for a in slots])) != len(slots):
			raise ValueError("slots' sql_names need to be unique: %s" % str([a.sql_name for a in slots]))

		if len(set([p.sql_name for p in processors])) != len(processors):
			raise ValueError("processors' sql_names need to be unique: %s" % str([p.sql_name for p in processors]))

		_check_disjunct_outputs(processors)

		for processor in processors:
			for x in processor.outputs:
				if x not in self._slots:
					raise ValueError("unknown slot '%s' referenced in output of processor '%s'" % (
						x, processor.name))
			for x in processor.inputs:
				if x not in self._slots:
					raise ValueError("unknown slot '%s' referenced in input of processor '%s'" % (
						x, processor.name))

		self._slot_sql_name = dict((a.name, a.sql_name) for a in slots)

		db_uri = 'sqlite:///%s' % str(Path(path))
		self._engine = sqlalchemy.create_engine(db_uri, isolation_level="SERIALIZABLE")

		# see https://docs.sqlalchemy.org/en/13/dialects/sqlite.html#pysqlite-serializable
		@sqlalchemy.event.listens_for(self._engine, "connect")
		def do_connect(dbapi_connection, connection_record):
		    # disable pysqlite's emitting of the BEGIN statement entirely.
		    # also stops it from emitting COMMIT before any DDL.
		    dbapi_connection.isolation_level = None

		@sqlalchemy.event.listens_for(self._engine, "begin")
		def do_begin(conn):
		    conn.execute("BEGIN EXCLUSIVE")

		metadata = sqlalchemy.MetaData(self._engine)

		self._task_table = self._create_task_table("task", metadata)

		self._data_table = dict()
		for a in slots:
			self._data_table[a.sql_name] = sqlalchemy.Table(SLOT_COL % a.sql_name, metadata,
				sqlalchemy.Column(
					'id', sqlalchemy.Integer, primary_key=True),
				sqlalchemy.Column(
					'data', sqlalchemy.LargeBinary if a.sql_type is None else a.sql_type))

		self._err_table = dict()
		for p in processors:
			self._err_table[p.sql_name] = sqlalchemy.Table('err_%s' % p.sql_name, metadata,
				sqlalchemy.Column(
					'id', sqlalchemy.Integer, primary_key=True),
				sqlalchemy.Column(
					'err', sqlalchemy.Text),
				extend_existing=True)

		metadata.create_all()

		self._migration = self._build_migration()

	def _create_task_table(self, name, metadata):
		tt = sqlalchemy.Table(name, metadata,
			sqlalchemy.Column(
				'id', sqlalchemy.Integer,
				sqlalchemy.Sequence('task_id_seq', metadata=metadata), primary_key=True),
			sqlalchemy.Column(
				'input', sqlalchemy.String,
				nullable=False, index=True, unique=True),
			*self._task_table_columns()
		)

		for p in self._processors.values():
			columns = [getattr(tt.c, PROCESS_STATE_COL % p.sql_name)] + [
				getattr(tt.c, SLOT_COL % self._slot_sql_name[x]) for x in p.inputs] + [
				getattr(tt.c, PROCESS_ITERATION_COL % p.sql_name)]
			sqlalchemy.Index('idx_proc_%s' % p, *columns)

		return tt

	def _task_table_columns(self):
		return list(itertools.chain(
			[sqlalchemy.Column(
				SLOT_COL % a.sql_name, 
				sqlalchemy.Boolean,
				nullable=False, default=False) for a in self._slots.values()],
			[sqlalchemy.Column(
				PROCESS_STATE_COL % proc.sql_name, 
				sqlalchemy.Enum(ProcessorState),
				nullable=False, default=ProcessorState.queued) for proc in self._processors.values()],
			[sqlalchemy.Column(
				PROCESS_ITERATION_COL % proc.sql_name, 
				sqlalchemy.Integer,
				nullable=False, default=0) for proc in self._processors.values()]))

	def _build_migration(self):
		migration = Migration(self, self._engine)
		if migration.is_empty:
			migration = None
		return migration

	def needs_migration(self):
		return self._migration is not None

	def _ensure_is_migrated(self):
		if self.needs_migration():
			raise NeedsMigration()

	def migrate(self):
		if self._migration:
			self._migration.run()
		sys.exit(0)

	@property
	def processors(self):
		return [p.name for p in self._processors.values()]

	def _slot_column_name(self, public_slot_name):
		return SLOT_COL % self._slot_sql_name[public_slot_name]

	def add_tasks(self, tasks, chunk_size=100, write_size=100, show_progress=False):
		self._ensure_is_migrated()

		tt = self._task_table
		all_processor_sql_names = set([p.sql_name for p in self._processors.values()])

		conn = self._engine.connect()

		try:

			with tqdm(total=len(tasks), desc='adding tasks', disable=not show_progress) as pbar:
				data = []

				for chunked_tasks in _chunks(tasks, chunk_size):

					by_input = dict((c.input_, c) for c in chunked_tasks)

					selection = sqlalchemy.sql.select([tt.c.input]).where(
						tt.c.input.in_([c.input_ for c in chunked_tasks]))
					existing_inputs = set(
						[row['input'] for row in conn.execute(selection).fetchall()])

					for task_input in set(by_input.keys()) - existing_inputs:
						task_data = dict(input=task_input)

						task = by_input[task_input]
						processor_names = task.processors
						if processor_names != "all":
							assert type(processor_names) is tuple

							ignored_processors = all_processor_sql_names - set([
								self._processors[p].sql_name for p in processor_names])
						else:
							ignored_processors = set()

						for p in all_processor_sql_names:
							state = ProcessorState.ignored if p in ignored_processors else ProcessorState.queued
							task_data[PROCESS_STATE_COL % p] = state

						data.append(task_data)

					if len(data) >= write_size:
						conn.execute(self._task_table.insert(), data)
						data = []

					pbar.update(len(chunked_tasks))

				if data:
					conn.execute(self._task_table.insert(), data)

		finally:
			conn.close()

	def _run_processor_core(self, conn, processor, task_id, task_input, task_stage, input_slots, interruptions):
		tt = self._task_table

		args = input_slots.copy()
		args['task_input'] = task_input

		inputs = collections.namedtuple('Inputs', ['task_input', *input_slots.keys()])(**args)

		outputs = recordclass.recordclass('Outputs', processor.outputs)(*[None for _ in processor.outputs])
		
		try:
			interruptions.allowed = True
			processor.run(inputs, outputs)
		finally:
			interruptions.allowed = False

		if not all(getattr(outputs, x) is not None for x in processor.outputs):
			raise ValueError("processor %s did not produce all required output slots" % processor.name)

		trans = conn.begin()
		try:
			for slot_name in processor.outputs:
				data_table = self._data_table[self._slot_sql_name[slot_name]]
				conn.execute(data_table.insert().values(
					id=task_id,
					data=getattr(outputs, slot_name)))

			new_state = dict()
			new_state[PROCESS_ITERATION_COL % processor.sql_name] = task_stage + 1

			if task_stage + 1 >= getattr(processor, 'n_stages', 1):
				new_state[PROCESS_STATE_COL % processor.sql_name] = ProcessorState.done

			for output in processor.outputs:
				new_state[self._slot_column_name(output)] = True

			stmt = tt.update().values(**new_state).where(tt.c.input == task_input)
			conn.execute(stmt)

			trans.commit()

		except:
			trans.rollback()
			raise

	def _load_slots(self, conn, slot_names, task_id):
		data = dict()
		for slot_name in slot_names:
			data_table = self._data_table[self._slot_sql_name[slot_name]]
			selection = sqlalchemy.sql.select([data_table.c.data]).where(
				data_table.c.id == task_id)
			data[slot_name] = conn.execute(selection).fetchone()['data']
		return data

	def _run_processor(self, conn, processor, show_progress, interruptions):
		tt = self._task_table
		any_work_done = False

		condition = sqlalchemy.and_(*[
			getattr(tt.c, self._slot_column_name(c)) == True for c in processor.inputs] + [
			getattr(tt.c, PROCESS_STATE_COL % processor.sql_name) == ProcessorState.queued])

		def query_remaining():
			selection = sqlalchemy.sql.select(
				[sqlalchemy.func.count()]).select_from(tt).where(condition)
			return conn.execute(selection).scalar()

		total = query_remaining()
		rem = total

		if total == 0:
			return False

		with tqdm(desc="running %s" % processor.name, disable=not show_progress, total=total) as pbar:

			while not interruptions.requested:

				trans = conn.begin()
				try:
					selection = sqlalchemy.sql.select([tt], for_update=True).where(condition).order_by(
						getattr(tt.c, PROCESS_ITERATION_COL % processor.sql_name))
					row = conn.execute(selection).fetchone()

					if row is None:  # done with this processor?
						trans.rollback()
						break

					any_work_done = True

					task_input = row[tt.c.input]
					task_id = row[tt.c.id]
					task_stage = row[getattr(tt.c, PROCESS_ITERATION_COL % processor.sql_name)]

					input_slots_data = self._load_slots(conn, processor.inputs, task_id)

					new_state = dict()
					new_state[PROCESS_STATE_COL % processor.sql_name] = ProcessorState.running

					stmt = tt.update().values(**new_state).where(tt.c.input == task_input)
					conn.execute(stmt)

					trans.commit()

				except:
					trans.rollback()
					raise

				try:
					self._run_processor_core(
						conn, processor, task_id, task_input, task_stage, input_slots_data, interruptions)

				except InterruptedException as e:

					trans = conn.begin()
					try:
						new_state = dict()
						new_state[PROCESS_STATE_COL % processor.sql_name] = ProcessorState.queued

						stmt = tt.update().values(**new_state).where(tt.c.input == task_input)
						conn.execute(stmt)

						trans.commit()

					except:
						trans.rollback()
						raise

					raise

				except:
					trans = conn.begin()
					try:
						err_table = self._err_table[processor.sql_name]

						stmt = err_table.delete().where(
							err_table.c.id == task_id)
						conn.execute(stmt)

						stmt = err_table.insert().values(
							id=task_id, err=traceback.format_exc())
						conn.execute(stmt)

						new_state = dict()
						new_state[PROCESS_STATE_COL % processor.sql_name] = ProcessorState.failed

						stmt = tt.update().values(**new_state).where(tt.c.input == task_input)
						conn.execute(stmt)

						trans.commit()

					except:
						trans.rollback()
						raise

					raise

				if show_progress:
					new_rem = query_remaining()
					pbar.update(rem - new_rem)
					rem = new_rem

		if interruptions.requested:
			raise InterruptedException()

		return any_work_done

	def _resolve_processors(self, processor_names):
		if processor_names == 'all':
			processor_names = list(self._processors.keys())

		return [self._processors[name] for name in processor_names]

	def run(self, processor_names='all', show_progress=False):
		self._ensure_is_migrated()
		processors = self._resolve_processors(processor_names)

		interruptions = Interruptions()
		def handler(signum, frame):
			interruptions.requested = True
			if interruptions.allowed:
				raise InterruptedException()

		old_handler = signal.signal(signal.SIGINT, handler)

		try:
			conn = self._engine.connect()

			try:
				while True:
					any_work_done = False

					for processor in processors:
						if self._run_processor(
							conn, processor, show_progress, interruptions):
							
							any_work_done = True

					if not any_work_done:
						break

			finally:
				conn.close()

		finally:
			signal.signal(signal.SIGINT, old_handler)

	def reset(self, processor_names='all', reset_hard=False):
		self._ensure_is_migrated()
		tt = self._task_table

		processors = self._resolve_processors(processor_names)

		if not reset_hard:
			selected_states = [ProcessorState.failed]
		else:
			selected_states = [ProcessorState.failed, ProcessorState.done, ProcessorState.ignored]

		conn = self._engine.connect()

		try:
			for processor in processors:

				condition = getattr(tt.c, PROCESS_STATE_COL % processor.sql_name).in_(selected_states)

				selection = sqlalchemy.sql.select([tt], for_update=True).where(condition)

				trans = conn.begin()
				try:
					for task in conn.execute(selection).fetchall():
						task_id = task['id']

						err_table = self._err_table[processor.sql_name]
						stmt = sqlalchemy.sql.delete(err_table).where(
							err_table.c.id == task_id)
						conn.execute(stmt)

						for x in processor.outputs:
							data_table = self._data_table[self._slot_sql_name[x]]
							stmt = sqlalchemy.sql.delete(data_table).where(
								data_table.c.id == task_id)
							conn.execute(stmt)

						new_state = dict()
						new_state[PROCESS_STATE_COL % processor.sql_name] = ProcessorState.queued

						stmt = tt.update().values(**new_state).where(tt.c.id == task_id)
						conn.execute(stmt)

					trans.commit()

				except:
					trans.rollback()
					raise

		finally:
			conn.close()		

	def print_status(self):
		self._ensure_is_migrated()

		tt = self._task_table
		status_table = []

		conn = self._engine.connect()

		try:
			for processor in sorted(self._processors.values(), key=lambda p: len(p.inputs)):
				counts = dict()

				state_c = getattr(tt.c, PROCESS_STATE_COL % processor.sql_name)
				selection = sqlalchemy.sql.select(
					[state_c, sqlalchemy.func.count()]).select_from(tt).group_by(state_c)
				for status in conn.execute(selection).fetchall():
					state, count = list(status.values())
					counts[state] = count

				status_table.append(
					[processor.name] + [counts.get(s, 0) for s in ProcessorState])

			print(tabulate(status_table, headers=["processor"] + [s.name for s in ProcessorState], tablefmt="psql"))

		finally:
			conn.close()

	def tasks(self):
		self._ensure_is_migrated()

		tt = self._task_table
		conn = self._engine.connect()

		try:
			selection = sqlalchemy.sql.select([tt])
			for task in conn.execute(selection).fetchall():

				processors = dict()
				for p in self._processors.values():
					processors[p.name] = task[PROCESS_STATE_COL % p.sql_name]

				slots = dict()
				for s in self._slots.values():
					slots[s.name] = task[SLOT_COL % s.sql_name]

				yield task['input'], processors, slots

		finally:
			conn.close()

	def print_tasks(self, up_to=10):
		self._ensure_is_migrated()

		processors = sorted(p.name for p in self._processors.values())
		slots = sorted(p.name for p in self._slots.values())

		tasks_table = []
		for task_input, task_processors, task_slots in itertools.islice(self.tasks(), 0, up_to):
			tasks_table.append(list(itertools.chain(
				[task_input],
				[task_processors[x].name for x in processors],
				[task_slots[x] for x in slots])))

		print(tabulate(tasks_table, headers=["input"] + processors + slots, tablefmt="psql"))

	def outputs(self, processor_name):
		self._ensure_is_migrated()
		processor = self._processors[processor_name]

		tt = self._task_table
		conn = self._engine.connect()

		try:
			condition = getattr(tt.c, PROCESS_STATE_COL % processor.sql_name) == ProcessorState.done

			selection = sqlalchemy.sql.select([tt]).where(condition)
			for task in conn.execute(selection).fetchall():
				yield task['input'], self._load_slots(conn, processor.outputs, task['id'])

		finally:
			conn.close()

	def print_outputs(self, processor_name, up_to=10):
		self._ensure_is_migrated()
		processor = self._processors[processor_name]
		outputs = processor.outputs

		outputs_table = []
		for task_input, output_data in itertools.islice(self.outputs(processor_name), 0, up_to):
			outputs_table.append([task_input] + [output_data[x] for x in outputs])

		print(tabulate(outputs_table, headers=["input"] + outputs, tablefmt="psql"))

	def slot(self, slot_name, task_input):
		self._ensure_is_migrated()
		slot = self._slots[slot_name]
		tt = self._task_table

		conn = self._engine.connect()

		try:
			selection = sqlalchemy.sql.select([tt]).where(tt.c.input == task_input)
			task = conn.execute(selection).fetchone()
			if task is None:
				raise ValueError("task '%s' not found" % task_input)

			dt = self._data_table[slot.sql_name]
			selection = sqlalchemy.sql.select([dt]).where(dt.c.id == task["id"])
			data = conn.execute(selection).fetchone()
			if data is None:
				raise ValueError("no data in slot '%s' for task '%s'" % (slot_name, task_input))

			return data["data"]

		finally:
			conn.close()


class Processor:
	def __init__(self, name, sql_name):
		self._name = name
		self._sql_name = sql_name

	def run(self, task, **kwargs):
		raise NotImplementedError()

	@property
	def name(self):
		return self._name

	@property
	def sql_name(self):
		return self._sql_name
# see https://docs.sqlalchemy.org/en/13/dialects/sqlite.html#pysqlite-serializable # disable pysqlite's emitting of the BEGIN statement entirely. # also stops it from emitting COMMIT before any DDL. # done with this processor?