Index name can be null.
Fixes: #1168
Change-Id: Id7c944e19a9facd7d3862d43f84fd70aedace999
or sqla_compat._constraint_is_named(c.const, autogen_context.dialect)
}
+ conn_uniques_by_name: Dict[sqla_compat._ConstraintName, _uq_constraint_sig]
+ conn_indexes_by_name: Dict[sqla_compat._ConstraintName, _ix_constraint_sig]
+
conn_uniques_by_name = {c.name: c for c in conn_unique_constraints}
- conn_indexes_by_name: Dict[Optional[str], _ix_constraint_sig] = {
- c.name: c for c in conn_indexes_sig
- }
+ conn_indexes_by_name = {c.name: c for c in conn_indexes_sig}
conn_names = {
c.name: c
- for c in conn_unique_constraints.union(
- conn_indexes_sig # type:ignore[arg-type]
- )
- if c.name is not None
+ for c in conn_unique_constraints.union(conn_indexes_sig)
+ if sqla_compat.constraint_name_defined(c.name)
}
doubled_constraints = {
from sqlalchemy import sql
from sqlalchemy import types as sqltypes
from sqlalchemy.sql.elements import conv
+from sqlalchemy.sql.elements import quoted_name
from .. import util
from ..operations import ops
from typing import Literal
from sqlalchemy.sql.elements import ColumnElement
- from sqlalchemy.sql.elements import quoted_name
from sqlalchemy.sql.elements import TextClause
from sqlalchemy.sql.schema import CheckConstraint
from sqlalchemy.sql.schema import Column
from sqlalchemy.sql.schema import Constraint
- from sqlalchemy.sql.schema import DefaultClause
from sqlalchemy.sql.schema import FetchedValue
from sqlalchemy.sql.schema import ForeignKey
from sqlalchemy.sql.schema import ForeignKeyConstraint
def _render_gen_name(
autogen_context: AutogenContext,
- name: Optional[Union[quoted_name, str]],
+ name: sqla_compat._ConstraintName,
) -> Optional[Union[quoted_name, str, _f_name]]:
if isinstance(name, conv):
return _f_name(_alembic_autogenerate_prefix(autogen_context), name)
else:
- return name
+ return sqla_compat.constraint_name_or_none(name)
def _indent(text: str) -> str:
"""
if name is None:
return name
- elif isinstance(name, sql.elements.quoted_name):
+ elif isinstance(name, quoted_name):
return str(name)
elif isinstance(name, str):
return name
}
-def _should_render_server_default_positionally(
- server_default: Union[Computed, DefaultClause]
-) -> bool:
+def _should_render_server_default_positionally(server_default: Any) -> bool:
return sqla_compat._server_default_is_computed(
server_default
) or sqla_compat._server_default_is_identity(server_default)
def __init__(
self,
- constraint_name: Optional[str],
+ constraint_name: sqla_compat._ConstraintName,
table_name: Union[str, quoted_name],
elements: Union[
Sequence[Tuple[str, str]],
cls, constraint: ExcludeConstraint
) -> CreateExcludeConstraintOp:
constraint_table = sqla_compat._table_for_constraint(constraint)
-
return cls(
constraint.name,
constraint_table.name,
(expr, op)
for expr, name, op in constraint._render_exprs # type:ignore[attr-defined] # noqa
],
- where=constraint.where,
+ where=cast(
+ "Optional[Union[BinaryExpression, str]]", constraint.where
+ ),
schema=constraint_table.schema,
_orig_constraint=constraint,
deferrable=constraint.deferrable,
"""
def create_index(
- index_name: str,
+ index_name: Optional[str],
table_name: str,
columns: Sequence[Union[str, TextClause, Function]],
schema: Optional[str] = None,
from ..util.sqla_compat import _remove_column_from_collection
from ..util.sqla_compat import _resolve_for_variant
from ..util.sqla_compat import _select
+from ..util.sqla_compat import constraint_name_defined
if TYPE_CHECKING:
from typing import Literal
# because
# we have no way to determine _is_type_bound() for these.
pass
- elif const.name:
+ elif constraint_name_defined(const.name):
self.named_constraints[const.name] = const
else:
self.unnamed_constraints.append(const)
"""
def add_constraint(self, const: Constraint) -> None:
- if not const.name:
+ if not constraint_name_defined(const.name):
raise ValueError("Constraint must have a name")
if isinstance(const, sql_schema.PrimaryKeyConstraint):
if self.table.primary_key in self.unnamed_constraints:
if col_const.name == const.name:
self.columns[col.name].constraints.remove(col_const)
else:
- assert const.name
+ assert constraint_name_defined(const.name)
const = self.named_constraints.pop(const.name)
except KeyError:
if _is_type_bound(const):
return ("remove_constraint", self.to_constraint())
@classmethod
- def from_constraint(
- cls,
- constraint: Constraint,
- ) -> DropConstraintOp:
+ def from_constraint(cls, constraint: Constraint) -> DropConstraintOp:
types = {
"unique_constraint": "unique",
"foreign_key_constraint": "foreignkey",
constraint_table = sqla_compat._table_for_constraint(constraint)
return cls(
- constraint.name,
+ sqla_compat.constraint_name_or_none(constraint.name),
constraint_table.name,
schema=constraint_table.schema,
type_=types[constraint.__visit_name__],
def from_constraint(cls, constraint: Constraint) -> CreatePrimaryKeyOp:
constraint_table = sqla_compat._table_for_constraint(constraint)
pk_constraint = cast("PrimaryKeyConstraint", constraint)
-
return cls(
- pk_constraint.name,
+ sqla_compat.constraint_name_or_none(pk_constraint.name),
constraint_table.name,
pk_constraint.columns.keys(),
schema=constraint_table.schema,
kw["initially"] = uq_constraint.initially
kw.update(uq_constraint.dialect_kwargs)
return cls(
- uq_constraint.name,
+ sqla_compat.constraint_name_or_none(uq_constraint.name),
constraint_table.name,
[c.name for c in uq_constraint.columns],
schema=constraint_table.schema,
kw["referent_schema"] = target_schema
kw.update(fk_constraint.dialect_kwargs)
return cls(
- fk_constraint.name,
+ sqla_compat.constraint_name_or_none(fk_constraint.name),
source_table,
target_table,
source_columns,
constraint_table = sqla_compat._table_for_constraint(constraint)
ck_constraint = cast("CheckConstraint", constraint)
-
return cls(
- ck_constraint.name,
+ sqla_compat.constraint_name_or_none(ck_constraint.name),
constraint_table.name,
cast("ColumnElement[Any]", ck_constraint.sqltext),
schema=constraint_table.schema,
def __init__(
self,
- index_name: str,
+ index_name: Optional[str],
table_name: str,
columns: Sequence[Union[str, TextClause, ColumnElement[Any]]],
schema: Optional[str] = None,
def create_index(
cls,
operations: Operations,
- index_name: str,
+ index_name: Optional[str],
table_name: str,
columns: Sequence[Union[str, TextClause, Function]],
schema: Optional[str] = None,
def index(
self,
- name: str,
+ name: Optional[str],
tablename: Optional[str],
columns: Sequence[Union[str, TextClause, ColumnElement[Any]]],
schema: Optional[str] = None,
from sqlalchemy.sql.elements import quoted_name
from sqlalchemy.sql.elements import TextClause
from sqlalchemy.sql.visitors import traverse
+from typing_extensions import TypeGuard
if TYPE_CHECKING:
from sqlalchemy import Index
_identity_attrs = _identity_options_attrs + ("on_null",)
has_identity = True
+if sqla_2:
+ from sqlalchemy.sql.base import _NoneName
+else:
+ from sqlalchemy.util import symbol as _NoneName # type: ignore[assignment]
+
+_ConstraintName = Union[None, str, _NoneName]
+
+
+def constraint_name_defined(name: _ConstraintName) -> TypeGuard[str]:
+ return isinstance(name, str)
+
+
+def constraint_name_or_none(name: _ConstraintName) -> Optional[str]:
+ return name if constraint_name_defined(name) else None
+
+
AUTOINCREMENT_DEFAULT = "auto"
Mako
importlib-metadata;python_version<"3.9"
importlib-resources;python_version<"3.9"
+ typing-extensions>=4
[options.extras_require]
tz =
basepython = python3
deps=
mypy
- sqlalchemy>=1.4.0
- sqlalchemy2-stubs
+ sqlalchemy>=2
mako
types-pkg-resources
types-python-dateutil
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
