schema: Optional[str],
tname: Union[quoted_name, str],
cname: Union[quoted_name, str],
- conn_col: Column,
- metadata_col: Column,
+ conn_col: Column[Any],
+ metadata_col: Column[Any],
) -> None:
metadata_col_nullable = metadata_col.nullable
schema: Optional[str],
tname: Union[quoted_name, str],
cname: quoted_name,
- conn_col: Column,
- metadata_col: Column,
+ conn_col: Column[Any],
+ metadata_col: Column[Any],
) -> None:
if metadata_col.table._autoincrement_column is metadata_col:
schema: Optional[str],
tname: Union[quoted_name, str],
cname: Union[quoted_name, str],
- conn_col: Column,
- metadata_col: Column,
+ conn_col: Column[Any],
+ metadata_col: Column[Any],
) -> None:
conn_type = conn_col.type
schema: Optional[str],
tname: str,
cname: str,
- conn_col: Column,
- metadata_col: Column,
+ conn_col: Column[Any],
+ metadata_col: Column[Any],
) -> None:
rendered_metadata_default = str(
cast(sa_schema.Computed, metadata_col.server_default).sqltext.compile(
schema: Optional[str],
tname: Union[quoted_name, str],
cname: Union[quoted_name, str],
- conn_col: Column,
- metadata_col: Column,
+ conn_col: Column[Any],
+ metadata_col: Column[Any],
) -> Optional[bool]:
metadata_default = metadata_col.server_default
schema: Optional[str],
tname: Union[quoted_name, str],
cname: quoted_name,
- conn_col: Column,
- metadata_col: Column,
+ conn_col: Column[Any],
+ metadata_col: Column[Any],
) -> Optional[Literal[False]]:
assert autogen_context.dialect is not None
return False
-def _render_column(column: Column, autogen_context: AutogenContext) -> str:
+def _render_column(
+ column: Column[Any], autogen_context: AutogenContext
+) -> str:
rendered = _user_defined_render("column", column, autogen_context)
if rendered is not False:
return rendered
def _render_server_default(
- default: Optional[Union[FetchedValue, str, TextClause, ColumnElement]],
+ default: Optional[
+ Union[FetchedValue, str, TextClause, ColumnElement[Any]]
+ ],
autogen_context: AutogenContext,
repr_: bool = True,
) -> Optional[str]:
Callable[
[
MigrationContext,
- Column,
- Column,
+ Column[Any],
+ Column[Any],
Optional[str],
Optional[FetchedValue],
Optional[str],
def __init__(
self,
name: str,
- column: Column,
+ column: Column[Any],
schema: Optional[Union[quoted_name, str]] = None,
) -> None:
super().__init__(name, schema=schema)
class DropColumn(AlterTable):
def __init__(
- self, name: str, column: Column, schema: Optional[str] = None
+ self, name: str, column: Column[Any], schema: Optional[str] = None
) -> None:
super().__init__(name, schema=schema)
self.column = column
return "ALTER COLUMN %s" % format_column_name(compiler, name)
-def add_column(compiler: DDLCompiler, column: Column, **kw) -> str:
+def add_column(compiler: DDLCompiler, column: Column[Any], **kw) -> str:
text = "ADD COLUMN %s" % compiler.get_column_specification(column, **kw)
const = " ".join(
def add_column(
self,
table_name: str,
- column: Column,
+ column: Column[Any],
schema: Optional[Union[str, quoted_name]] = None,
) -> None:
self._exec(base.AddColumn(table_name, column, schema=schema))
def drop_column(
self,
table_name: str,
- column: Column,
+ column: Column[Any],
schema: Optional[str] = None,
**kw,
) -> None:
def drop_table_comment(self, table: Table) -> None:
self._exec(schema.DropTableComment(table))
- def create_column_comment(self, column: ColumnElement) -> None:
+ def create_column_comment(self, column: ColumnElement[Any]) -> None:
self._exec(schema.SetColumnComment(column))
def drop_index(self, index: Index) -> None:
return True
def compare_type(
- self, inspector_column: Column, metadata_column: Column
+ self, inspector_column: Column[Any], metadata_column: Column
) -> bool:
"""Returns True if there ARE differences between the types of the two
columns. Takes impl.type_synonyms into account between retrospected
def drop_column(
self,
table_name: str,
- column: Column,
+ column: Column[Any],
schema: Optional[str] = None,
**kw,
) -> None:
def __init__(
self,
tname: str,
- colname: Union[Column, str],
+ colname: Union[Column[Any], str],
type_: str,
schema: Optional[str],
) -> None:
inherit_cache = False
def __init__(
- self, tname: str, colname: Column, schema: Optional[str]
+ self, tname: str, colname: Column[Any], schema: Optional[str]
) -> None:
self.tname = tname
self.colname = colname
)
-def mssql_add_column(compiler: MSDDLCompiler, column: Column, **kw) -> str:
+def mssql_add_column(
+ compiler: MSDDLCompiler, column: Column[Any], **kw
+) -> str:
return "ADD %s" % compiler.get_column_specification(column, **kw)
return "MODIFY %s" % format_column_name(compiler, name)
-def add_column(compiler: OracleDDLCompiler, column: Column, **kw) -> str:
+def add_column(compiler: OracleDDLCompiler, column: Column[Any], **kw) -> str:
return "ADD %s" % compiler.get_column_specification(column, **kw)
table_name: Union[str, quoted_name],
elements: Union[
Sequence[Tuple[str, str]],
- Sequence[Tuple[ColumnClause, str]],
+ Sequence[Tuple[ColumnClause[Any], str]],
],
where: Optional[Union[BinaryExpression, str]] = None,
schema: Optional[str] = None,
def _render_potential_column(
- value: Union[ColumnClause, Column, TextClause, FunctionElement],
+ value: Union[
+ ColumnClause[Any], Column[Any], TextClause, FunctionElement[Any]
+ ],
autogen_context: AutogenContext,
) -> str:
if isinstance(value, ColumnClause):
def compare_server_default(
self,
- inspector_column: Column,
- metadata_column: Column,
+ inspector_column: Column[Any],
+ metadata_column: Column[Any],
rendered_metadata_default: Optional[str],
rendered_inspector_default: Optional[str],
) -> bool:
def cast_for_batch_migrate(
self,
- existing: Column,
+ existing: Column[Any],
existing_transfer: Dict[str, Union[TypeEngine, Cast]],
new_type: TypeEngine,
) -> None:
### end imports ###
def add_column(
- table_name: str, column: Column, *, schema: Optional[str] = None
+ table_name: str, column: Column[Any], *, schema: Optional[str] = None
) -> None:
"""Issue an "add column" instruction using the current
migration context.
def add_column(
self,
table_name: str,
- column: Column,
+ column: Column[Any],
*,
schema: Optional[str] = None,
) -> None:
def add_column(
self,
- column: Column,
+ column: Column[Any],
*,
insert_before: Optional[str] = None,
insert_after: Optional[str] = None,
def _grab_table_elements(self) -> None:
schema = self.table.schema
- self.columns: Dict[str, Column] = OrderedDict()
+ self.columns: Dict[str, Column[Any]] = OrderedDict()
for c in self.table.c:
c_copy = _copy(c, schema=schema)
c_copy.unique = c_copy.index = False
def add_column(
self,
table_name: str,
- column: Column,
+ column: Column[Any],
insert_before: Optional[str] = None,
insert_after: Optional[str] = None,
**kw,
self.column_transfers[column.name] = {}
def drop_column(
- self, table_name: str, column: Union[ColumnClause, Column], **kw
+ self,
+ table_name: str,
+ column: Union[ColumnClause[Any], Column[Any]],
+ **kw,
) -> None:
if column.name in self.table.primary_key.columns:
_remove_column_from_collection(
def __init__(
self,
table_name: str,
- column: Column,
+ column: Column[Any],
*,
schema: Optional[str] = None,
**kw: Any,
def to_diff_tuple(
self,
- ) -> Tuple[str, Optional[str], str, Column]:
+ ) -> Tuple[str, Optional[str], str, Column[Any]]:
return ("add_column", self.schema, self.table_name, self.column)
def to_column(self) -> Column:
cls,
schema: Optional[str],
tname: str,
- col: Column,
+ col: Column[Any],
) -> AddColumnOp:
return cls(tname, col, schema=schema)
cls,
operations: Operations,
table_name: str,
- column: Column,
+ column: Column[Any],
*,
schema: Optional[str] = None,
) -> None:
def batch_add_column(
cls,
operations: BatchOperations,
- column: Column,
+ column: Column[Any],
*,
insert_before: Optional[str] = None,
insert_after: Optional[str] = None,
def to_diff_tuple(
self,
- ) -> Tuple[str, Optional[str], str, Column]:
+ ) -> Tuple[str, Optional[str], str, Column[Any]]:
return (
"remove_column",
self.schema,
cls,
schema: Optional[str],
tname: str,
- col: Column,
+ col: Column[Any],
) -> DropColumnOp:
return cls(
tname,
CompareServerDefault = Callable[
[
MigrationContext,
- Column,
- Column,
+ "Column[Any]",
+ "Column[Any]",
Optional[str],
Optional[FetchedValue],
Optional[str],
return None
def _compare_type(
- self, inspector_column: Column, metadata_column: Column
+ self, inspector_column: Column[Any], metadata_column: Column
) -> bool:
if self._user_compare_type is False:
return False
def _compare_server_default(
self,
- inspector_column: Column,
- metadata_column: Column,
+ inspector_column: Column[Any],
+ metadata_column: Column[Any],
rendered_metadata_default: Optional[str],
rendered_column_default: Optional[str],
) -> bool:
from sqlalchemy.sql.selectable import Select
from sqlalchemy.sql.selectable import TableClause
-_CE = TypeVar("_CE", bound=Union["ColumnElement", "SchemaItem"])
+_CE = TypeVar("_CE", bound=Union["ColumnElement[Any]", "SchemaItem"])
def _safe_int(value: str) -> Union[int, str]:
def _remove_column_from_collection(
- collection: ColumnCollection, column: Union[Column, ColumnClause]
+ collection: ColumnCollection, column: Union[Column[Any], ColumnClause[Any]]
) -> None:
"""remove a column from a ColumnCollection."""
def _textual_index_column(
- table: Table, text_: Union[str, TextClause, ColumnElement]
-) -> Union[ColumnElement, Column]:
+ table: Table, text_: Union[str, TextClause, ColumnElement[Any]]
+) -> Union[ColumnElement[Any], Column[Any]]:
"""a workaround for the Index construct's severe lack of flexibility"""
if isinstance(text_, str):
c = Column(text_, sqltypes.NULLTYPE)
--- /dev/null
+.. change::
+ :tags: bug, typing
+ :tickets: 1246
+
+ Fixed typing use of :class:`~sqlalchemy.schema.Column` and other
+ generic SQLAlchemy classes.
"sqlalchemy.sql.type_api.",
"sqlalchemy.sql.functions.",
"sqlalchemy.sql.dml.",
+ "typing."
]
ADDITIONAL_ENV = {
"MigrationContext": MigrationContext,
retval = annotation.__qualname__
elif isinstance(annotation, typing.TypeVar):
retval = annotation.__name__
+ elif hasattr(annotation, "__args__") and hasattr(
+ annotation, "__origin__"
+ ):
+ # generic class
+ retval = str(annotation)
else:
retval = annotation