Remove class InsertFromSelect

This class in new edtion is useless.
So class InsertFromSelect is
deprecated for removal.

Change-Id: I200486a5a88e78b5223b9364fbea5901048d5cb9
Closes-Bug:#1714768
This commit is contained in:
zhangyangyang 2017-09-03 20:00:24 +08:00
parent e9a9701d54
commit 3d8fe3e12b
2 changed files with 0 additions and 158 deletions
oslo_db
sqlalchemy
tests/sqlalchemy

@ -22,7 +22,6 @@ import itertools
import logging
import re
import debtcollector
from oslo_utils import timeutils
import six
import sqlalchemy
@ -433,43 +432,6 @@ def get_table(engine, name):
return Table(name, metadata, autoload=True)
@debtcollector.removals.removed_class(
'InsertFromSelect',
replacement='sqlalchemy.sql.expression.Insert.from_select',
message='this functionality is provided out-of-box by SQLAlchemy >= 1.0.0'
)
class InsertFromSelect(object):
"""Form the base for `INSERT INTO table (SELECT ... )` statement.
:param table: table to insert records
:param select: select query
:param cols: list of columns to specify in insert clause
:return: SQLAlchemy :class:`Insert` object instance
Usage:
.. code-block:: python
select = sql.select(table_from)
insert = InsertFromSelect(table_to, select,
['id', 'name', 'insert_date'])
engine.execute(insert)
"""
# NOTE(tdurakov): Insert from select implementation added to SQLAlchemy
# starting from version 0.8.7. Default SQLAlchemy implementation should be
# used instead of this. Deprecated.
def __new__(cls, table, select, cols=None):
if not cols:
cols = [c.name for c in table.c]
return table.insert(inline=True).from_select(cols, select)
def __init__(self, table, select, cols=None):
pass
def _get_not_supported_column(col_name_col_instance, column_name):
try:
column = col_name_col_instance[column_name]
@ -628,9 +590,6 @@ def _change_deleted_column_type_to_boolean_sqlite(engine, table_name,
else:
c_select.append(table.c.deleted == table.c.id)
ins = InsertFromSelect(new_table, sqlalchemy.sql.select(c_select))
engine.execute(ins)
table.drop()
for index in indexes:
index.create(engine)
@ -722,9 +681,6 @@ def _change_deleted_column_type_to_id_type_sqlite(engine, table_name,
indexes.append(Index(index["name"], *column_names,
unique=index["unique"]))
ins = InsertFromSelect(new_table, table.select())
engine.execute(ins)
table.drop()
for index in indexes:
index.create(engine)

@ -15,7 +15,6 @@
import fixtures
import mock
from oslo_utils import uuidutils
from oslotest import base as test_base
from oslotest import moxstubout
from six.moves.urllib import parse
@ -771,119 +770,6 @@ class TestMigrationUtils(db_test_base.DbTestCase):
# behavior), any integer value can be inserted, otherwise only 1 or 0.
self.engine.execute(table.insert({'deleted': 10}))
def test_insert_from_select(self):
insert_table_name = "__test_insert_to_table__"
select_table_name = "__test_select_from_table__"
uuidstrs = []
for unused in range(10):
uuidstrs.append(uuidutils.generate_uuid(dashed=False))
insert_table = Table(
insert_table_name, self.meta,
Column('id', Integer, primary_key=True,
nullable=False, autoincrement=True),
Column('uuid', String(36), nullable=False))
select_table = Table(
select_table_name, self.meta,
Column('id', Integer, primary_key=True,
nullable=False, autoincrement=True),
Column('uuid', String(36), nullable=False))
insert_table.create()
select_table.create()
# Add 10 rows to select_table
for uuidstr in uuidstrs:
ins_stmt = select_table.insert().values(uuid=uuidstr)
self.conn.execute(ins_stmt)
# Select 4 rows in one chunk from select_table
column = select_table.c.id
query_insert = select([select_table],
select_table.c.id < 5).order_by(column)
insert_statement = utils.InsertFromSelect(insert_table,
query_insert)
result_insert = self.conn.execute(insert_statement)
# Verify we insert 4 rows
self.assertEqual(4, result_insert.rowcount)
query_all = select([insert_table]).where(
insert_table.c.uuid.in_(uuidstrs))
rows = self.conn.execute(query_all).fetchall()
# Verify we really have 4 rows in insert_table
self.assertEqual(4, len(rows))
def test_insert_from_select_with_specified_columns(self):
insert_table_name = "__test_insert_to_table__"
select_table_name = "__test_select_from_table__"
uuidstrs = []
for unused in range(10):
uuidstrs.append(uuidutils.generate_uuid(dashed=False))
insert_table = Table(
insert_table_name, self.meta,
Column('id', Integer, primary_key=True,
nullable=False, autoincrement=True),
Column('uuid', String(36), nullable=False))
select_table = Table(
select_table_name, self.meta,
Column('id', Integer, primary_key=True,
nullable=False, autoincrement=True),
Column('uuid', String(36), nullable=False))
insert_table.create()
select_table.create()
# Add 10 rows to select_table
for uuidstr in uuidstrs:
ins_stmt = select_table.insert().values(uuid=uuidstr)
self.conn.execute(ins_stmt)
# Select 4 rows in one chunk from select_table
column = select_table.c.id
query_insert = select([select_table],
select_table.c.id < 5).order_by(column)
insert_statement = utils.InsertFromSelect(insert_table,
query_insert, ['id', 'uuid'])
result_insert = self.conn.execute(insert_statement)
# Verify we insert 4 rows
self.assertEqual(4, result_insert.rowcount)
query_all = select([insert_table]).where(
insert_table.c.uuid.in_(uuidstrs))
rows = self.conn.execute(query_all).fetchall()
# Verify we really have 4 rows in insert_table
self.assertEqual(4, len(rows))
def test_insert_from_select_with_specified_columns_negative(self):
insert_table_name = "__test_insert_to_table__"
select_table_name = "__test_select_from_table__"
uuidstrs = []
for unused in range(10):
uuidstrs.append(uuidutils.generate_uuid(dashed=False))
insert_table = Table(
insert_table_name, self.meta,
Column('id', Integer, primary_key=True,
nullable=False, autoincrement=True),
Column('uuid', String(36), nullable=False))
select_table = Table(
select_table_name, self.meta,
Column('id', Integer, primary_key=True,
nullable=False, autoincrement=True),
Column('uuid', String(36), nullable=False))
insert_table.create()
select_table.create()
# Add 10 rows to select_table
for uuidstr in uuidstrs:
ins_stmt = select_table.insert().values(uuid=uuidstr)
self.conn.execute(ins_stmt)
# Select 4 rows in one chunk from select_table
column = select_table.c.id
query_insert = select([select_table],
select_table.c.id < 5).order_by(column)
insert_statement = utils.InsertFromSelect(insert_table,
query_insert, ['uuid', 'id'])
self.assertRaises(exception.DBError, self.conn.execute,
insert_statement)
class PostgesqlTestMigrations(TestMigrationUtils,
db_test_base.PostgreSQLOpportunisticTestCase):