def create_table(self):
if not self.schema:
self.schema = DB_ETL_SCHEMA
if not self.create_schema():
return False
logger.info('try to create table {} in {}'.format(
self.sql_table_name,
self.schema
))
if self.exist_table():
return True
table = self.get_sql_table_object(need_columns=True)
db_table = self.local_engine.execute(CreateTable(table))
for index in table.indexes:
self.local_engine.execute(CreateIndex(index))
return db_table
python类CreateIndex()的实例源码
def print_create_table(tables):
app.config.from_object('config.default')
database.init_app(app)
engine = database.session.get_bind()
for class_name in tables:
cls = get_class(class_name)
for c in cls.__table__.columns:
if not isinstance(c.type, Enum):
continue
t = c.type
sql = str(CreateEnumType(t).compile(engine))
click.echo(sql.strip() + ';')
for index in cls.__table__.indexes:
sql = str(CreateIndex(index).compile(engine))
click.echo(sql.strip() + ';')
sql = str(CreateTable(cls.__table__).compile(engine))
click.echo(sql.strip() + ';')
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))
def create_table(self, table):
if util.sqla_07:
table.dispatch.before_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
self._exec(schema.CreateTable(table))
if util.sqla_07:
table.dispatch.after_create(table, self.connection,
checkfirst=False,
_ddl_runner=self)
for index in table.indexes:
self._exec(schema.CreateIndex(index))
def create_index(self, index):
self._exec(schema.CreateIndex(index))