我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用sqlalchemy.exc.NoSuchTableError()。
def get_table_id(self, connection, table_name, schema=None, **kw): """Fetch the id for schema.table_name. Several reflection methods require the table id. The idea for using this method is that it can be fetched one time and cached for subsequent calls. """ table_id = None if schema is None: schema = self.default_schema_name TABLEID_SQL = text(""" SELECT o.id AS id FROM sysobjects o JOIN sysusers u ON o.uid=u.uid WHERE u.name = :schema_name AND o.name = :table_name AND o.type in ('U', 'V') """) if util.py2k: if isinstance(schema, unicode): schema = schema.encode("ascii") if isinstance(table_name, unicode): table_name = table_name.encode("ascii") result = connection.execute(TABLEID_SQL, schema_name=schema, table_name=table_name) table_id = result.scalar() if table_id is None: raise exc.NoSuchTableError(table_name) return table_id
def has_table(self, connection, table_name, schema=None): try: self.get_table_id(connection, table_name, schema) except exc.NoSuchTableError: return False else: return True
def has_table(self, connection, table_name, schema=None): try: self.get_columns(connection, table_name, schema) return True except NoSuchTableError: return False
def get_table_id(self, connection, table_name, schema=None, **kw): """Fetch the id for schema.table_name. Several reflection methods require the table id. The idea for using this method is that it can be fetched one time and cached for subsequent calls. """ table_id = None if schema is None: schema = self.default_schema_name TABLEID_SQL = text(""" SELECT o.id AS id FROM sysobjects o JOIN sysusers u ON o.uid=u.uid WHERE u.name = :schema_name AND o.name = :table_name AND o.type in ('U', 'V') """) # Py2K if isinstance(schema, unicode): schema = schema.encode("ascii") if isinstance(table_name, unicode): table_name = table_name.encode("ascii") # end Py2K result = connection.execute(TABLEID_SQL, schema_name=schema, table_name=table_name) table_id = result.scalar() if table_id is None: raise exc.NoSuchTableError(table_name) return table_id
def has_table(self, connection, table_name, schema=None): try: self._get_table_columns(connection, table_name, schema) return True except exc.NoSuchTableError: return False
def _get_table(self, connection, table_name, schema=None): if isinstance(connection, Engine): connection = connection.connect() project, dataset, table_name_prepared = self._split_table_name(table_name) if dataset is None and schema is not None: dataset = schema table_name_prepared = table_name table = connection.connection._client.dataset(dataset, project=project).table(table_name_prepared) try: t = connection.connection._client.get_table(table) except NotFound as e: raise NoSuchTableError(table_name) return t
def has_table(self, connection, table_name, schema=None): try: self._get_table(connection, table_name, schema) return True except NoSuchTableError: return False
def test_reflect_table_does_not_exist(engine): with pytest.raises(NoSuchTableError): table = Table('test_pybigquery.table_does_not_exist', MetaData(bind=engine), autoload=True) assert Table('test_pybigquery.table_does_not_exist', MetaData(bind=engine)).exists() is False
def test_reflect_dataset_does_not_exist(engine): with pytest.raises(NoSuchTableError): Table('dataset_does_not_exist.table_does_not_exist', MetaData(bind=engine), autoload=True)