def correct_for_autogen_constraints(self, conn_unique_constraints,
conn_indexes,
metadata_unique_constraints,
metadata_indexes):
conn_uniques_by_name = dict(
(c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict(
(c.name, c) for c in conn_indexes)
# TODO: if SQLA 1.0, make use of "duplicates_constraint"
# metadata
doubled_constraints = dict(
(name, (conn_uniques_by_name[name], conn_indexes_by_name[name]))
for name in set(conn_uniques_by_name).intersection(
conn_indexes_by_name)
)
for name, (uq, ix) in doubled_constraints.items():
conn_indexes.remove(ix)
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
if compat.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
for expr in exprs:
if not isinstance(expr, (Column, UnaryExpression)):
util.warn(
"autogenerate skipping functional index %s; "
"not supported by SQLAlchemy reflection" % idx.name
)
metadata_indexes.discard(idx)
python类UnaryExpression()的实例源码
def __init__(self, x):
if isinstance(x, unicode):
x = column(x)
if not isinstance(x, UnaryExpression):
x = asc(x)
self.uo = x
self.full_name = str(self.element)
try:
table_name, name = self.full_name.split('.', 1)
except ValueError:
table_name = None
name = self.full_name
self.table_name = table_name
self.name = name
def element(self):
x = self.uo
while isinstance(x, UnaryExpression):
x = x.element
return x
def is_ascending(self):
x = self.uo
while isinstance(x, UnaryExpression):
if x.modifier in (asc_op, desc_op):
return x.modifier == asc_op
else:
x = x.element
raise ValueError # pragma: no cover
def reversed(self):
x = copied = copy(self.uo)
while isinstance(x, UnaryExpression):
if x.modifier in (asc_op, desc_op):
if x.modifier == asc_op:
x.modifier = desc_op
else:
x.modifier = asc_op
return OC(copied)
else:
x = x.element
raise ValueError # pragma: no cover
def correct_for_autogen_constraints(self, conn_unique_constraints,
conn_indexes,
metadata_unique_constraints,
metadata_indexes):
conn_uniques_by_name = dict(
(c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict(
(c.name, c) for c in conn_indexes)
# TODO: if SQLA 1.0, make use of "duplicates_constraint"
# metadata
doubled_constraints = dict(
(name, (conn_uniques_by_name[name], conn_indexes_by_name[name]))
for name in set(conn_uniques_by_name).intersection(
conn_indexes_by_name)
)
for name, (uq, ix) in doubled_constraints.items():
conn_indexes.remove(ix)
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
if compat.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
for expr in exprs:
if not isinstance(expr, (Column, UnaryExpression)):
util.warn(
"autogenerate skipping functional index %s; "
"not supported by SQLAlchemy reflection" % idx.name
)
metadata_indexes.discard(idx)
def correct_for_autogen_constraints(self, conn_unique_constraints,
conn_indexes,
metadata_unique_constraints,
metadata_indexes):
conn_uniques_by_name = dict(
(c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict(
(c.name, c) for c in conn_indexes)
# TODO: if SQLA 1.0, make use of "duplicates_constraint"
# metadata
doubled_constraints = dict(
(name, (conn_uniques_by_name[name], conn_indexes_by_name[name]))
for name in set(conn_uniques_by_name).intersection(
conn_indexes_by_name)
)
for name, (uq, ix) in doubled_constraints.items():
conn_indexes.remove(ix)
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
if util.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
for expr in exprs:
while isinstance(expr, UnaryExpression):
expr = expr.element
if not isinstance(expr, Column):
util.warn(
"autogenerate skipping functional index %s; "
"not supported by SQLAlchemy reflection" % idx.name
)
metadata_indexes.discard(idx)
def correct_for_autogen_constraints(self, conn_unique_constraints,
conn_indexes,
metadata_unique_constraints,
metadata_indexes):
conn_uniques_by_name = dict(
(c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict(
(c.name, c) for c in conn_indexes)
# TODO: if SQLA 1.0, make use of "duplicates_constraint"
# metadata
doubled_constraints = dict(
(name, (conn_uniques_by_name[name], conn_indexes_by_name[name]))
for name in set(conn_uniques_by_name).intersection(
conn_indexes_by_name)
)
for name, (uq, ix) in doubled_constraints.items():
conn_indexes.remove(ix)
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
if compat.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
for expr in exprs:
if not isinstance(expr, (Column, UnaryExpression)):
util.warn(
"autogenerate skipping functional index %s; "
"not supported by SQLAlchemy reflection" % idx.name
)
metadata_indexes.discard(idx)
def correct_for_autogen_constraints(self, conn_unique_constraints,
conn_indexes,
metadata_unique_constraints,
metadata_indexes):
conn_uniques_by_name = dict(
(c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict(
(c.name, c) for c in conn_indexes)
# TODO: if SQLA 1.0, make use of "duplicates_constraint"
# metadata
doubled_constraints = dict(
(name, (conn_uniques_by_name[name], conn_indexes_by_name[name]))
for name in set(conn_uniques_by_name).intersection(
conn_indexes_by_name)
)
for name, (uq, ix) in doubled_constraints.items():
conn_indexes.remove(ix)
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
if compat.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
for expr in exprs:
if not isinstance(expr, (Column, UnaryExpression)):
util.warn(
"autogenerate skipping functional index %s; "
"not supported by SQLAlchemy reflection" % idx.name
)
metadata_indexes.discard(idx)
def correct_for_autogen_constraints(self, conn_unique_constraints,
conn_indexes,
metadata_unique_constraints,
metadata_indexes):
conn_uniques_by_name = dict(
(c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict(
(c.name, c) for c in conn_indexes)
# TODO: if SQLA 1.0, make use of "duplicates_constraint"
# metadata
doubled_constraints = dict(
(name, (conn_uniques_by_name[name], conn_indexes_by_name[name]))
for name in set(conn_uniques_by_name).intersection(
conn_indexes_by_name)
)
for name, (uq, ix) in doubled_constraints.items():
conn_indexes.remove(ix)
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
if compat.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
for expr in exprs:
while isinstance(expr, UnaryExpression):
expr = expr.element
if not isinstance(expr, Column):
util.warn(
"autogenerate skipping functional index %s; "
"not supported by SQLAlchemy reflection" % idx.name
)
metadata_indexes.discard(idx)
def correct_for_autogen_constraints(self, conn_unique_constraints,
conn_indexes,
metadata_unique_constraints,
metadata_indexes):
conn_uniques_by_name = dict(
(c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict(
(c.name, c) for c in conn_indexes)
# TODO: if SQLA 1.0, make use of "duplicates_constraint"
# metadata
doubled_constraints = dict(
(name, (conn_uniques_by_name[name], conn_indexes_by_name[name]))
for name in set(conn_uniques_by_name).intersection(
conn_indexes_by_name)
)
for name, (uq, ix) in doubled_constraints.items():
conn_indexes.remove(ix)
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
if util.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
for expr in exprs:
while isinstance(expr, UnaryExpression):
expr = expr.element
if not isinstance(expr, Column):
util.warn(
"autogenerate skipping functional index %s; "
"not supported by SQLAlchemy reflection" % idx.name
)
metadata_indexes.discard(idx)
def correct_for_autogen_constraints(self, conn_unique_constraints,
conn_indexes,
metadata_unique_constraints,
metadata_indexes):
conn_uniques_by_name = dict(
(c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict(
(c.name, c) for c in conn_indexes)
# TODO: if SQLA 1.0, make use of "duplicates_constraint"
# metadata
doubled_constraints = dict(
(name, (conn_uniques_by_name[name], conn_indexes_by_name[name]))
for name in set(conn_uniques_by_name).intersection(
conn_indexes_by_name)
)
for name, (uq, ix) in doubled_constraints.items():
conn_indexes.remove(ix)
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
if util.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
for expr in exprs:
while isinstance(expr, UnaryExpression):
expr = expr.element
if not isinstance(expr, Column):
util.warn(
"autogenerate skipping functional index %s; "
"not supported by SQLAlchemy reflection" % idx.name
)
metadata_indexes.discard(idx)
def correct_for_autogen_constraints(self, conn_unique_constraints,
conn_indexes,
metadata_unique_constraints,
metadata_indexes):
conn_uniques_by_name = dict(
(c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict(
(c.name, c) for c in conn_indexes)
# TODO: if SQLA 1.0, make use of "duplicates_constraint"
# metadata
doubled_constraints = dict(
(name, (conn_uniques_by_name[name], conn_indexes_by_name[name]))
for name in set(conn_uniques_by_name).intersection(
conn_indexes_by_name)
)
for name, (uq, ix) in doubled_constraints.items():
conn_indexes.remove(ix)
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
if compat.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
for expr in exprs:
while isinstance(expr, UnaryExpression):
expr = expr.element
if not isinstance(expr, Column):
util.warn(
"autogenerate skipping functional index %s; "
"not supported by SQLAlchemy reflection" % idx.name
)
metadata_indexes.discard(idx)
def correct_for_autogen_constraints(self, conn_unique_constraints,
conn_indexes,
metadata_unique_constraints,
metadata_indexes):
conn_uniques_by_name = dict(
(c.name, c) for c in conn_unique_constraints)
conn_indexes_by_name = dict(
(c.name, c) for c in conn_indexes)
# TODO: if SQLA 1.0, make use of "duplicates_constraint"
# metadata
doubled_constraints = dict(
(name, (conn_uniques_by_name[name], conn_indexes_by_name[name]))
for name in set(conn_uniques_by_name).intersection(
conn_indexes_by_name)
)
for name, (uq, ix) in doubled_constraints.items():
conn_indexes.remove(ix)
for idx in list(metadata_indexes):
if idx.name in conn_indexes_by_name:
continue
if compat.sqla_08:
exprs = idx.expressions
else:
exprs = idx.columns
for expr in exprs:
if not isinstance(expr, (Column, UnaryExpression)):
util.warn(
"autogenerate skipping functional index %s; "
"not supported by SQLAlchemy reflection" % idx.name
)
metadata_indexes.discard(idx)
