From a CTL model to select the scan chain from sqlite
TODO List
Create, insert, query into sqlite database
Trace the raw sql statement
Support included from ctl file or manually from shell or script
Mysql not support yet
Migrate database between sqlite and mysql
Change history
Add database type convert factory mode for sql insert, create
def_create_adapter(self, statement, table_type):
""" Factory to create raw sql create table statement base on the database type"""iftable_type=="sqlite":
raw_create_sql="CREATE TABLE %s (%s)"% (self.__table__, ",".join(statement))
eliftable_type=="mysql":
# TODOraiseNotImplementedError("Mysql database still not be supported")
else:
raiseNotImplementedError("Other database still not be supported")
returnraw_create_sqldef__init__(self):
""" Need to implement the Base class __init__ function. """super(self.__class__, self).__init__()
# Adapter to the table type implemented in the class. Default is sqliteself._create_adapter=partial(self._create_adapter, table_type=self.__table_type__)
self._insert_adapter=partial(self._insert_adapter, table_type=self.__table_type__)
Add context manager help function for connection of database
@contextmanagerdefconnect_hold(database_conn):
""" a context to hold the connection of database, like 'with connect_hold(**) as c' """c=database_conn.cursor()
try:
yieldcfinally:
c.close()
database_conn.commit()
database_conn.close()
defget_database_connection(database_name, database_type):
"""Return connection from sqlite database"""ifdatabase_type!="sqlite":
raiseNotImplementedError("Only sqlite database is supported for now")
returnsqlite3.connect(database_name)
@classmethoddefraw_sql(cls, sql_string):
""" execute the raw sql statement in sql_string """withconnect_hold(get_database_connection(cls.__table__+".db", "sqlite")) asc:
c.execute(sql_string)
returnc.fetchall()