# mypy: ignore-errors
from __future__ import annotations
+from types import ModuleType
from typing import Any
+from typing import Dict
from typing import Optional
from typing import Type
from typing import TYPE_CHECKING
from ...sql.ddl import InvokeDropDDLBase
if TYPE_CHECKING:
+ from ...sql._typing import _CreateDropBind
from ...sql._typing import _TypeEngineArgument
-class NamedType(sqltypes.TypeEngine):
+class NamedType(schema.SchemaVisitable, sqltypes.TypeEngine):
"""Base for named types."""
__abstract__ = True
DDLDropper: Type[NamedTypeDropper]
create_type: bool
- def create(self, bind, checkfirst=True, **kw):
+ def create(
+ self, bind: _CreateDropBind, checkfirst: bool = True, **kw: Any
+ ) -> None:
"""Emit ``CREATE`` DDL for this type.
:param bind: a connectable :class:`_engine.Engine`,
"""
bind._run_ddl_visitor(self.DDLGenerator, self, checkfirst=checkfirst)
- def drop(self, bind, checkfirst=True, **kw):
+ def drop(
+ self, bind: _CreateDropBind, checkfirst: bool = True, **kw: Any
+ ) -> None:
"""Emit ``DROP`` DDL for this type.
:param bind: a connectable :class:`_engine.Engine`,
"""
bind._run_ddl_visitor(self.DDLDropper, self, checkfirst=checkfirst)
- def _check_for_name_in_memos(self, checkfirst, kw):
+ def _check_for_name_in_memos(
+ self, checkfirst: bool, kw: Dict[str, Any]
+ ) -> bool:
"""Look in the 'ddl runner' for 'memos', then
note our name in that collection.
else:
return False
- def _on_table_create(self, target, bind, checkfirst=False, **kw):
+ def _on_table_create(
+ self,
+ target: Any,
+ bind: _CreateDropBind,
+ checkfirst: bool = False,
+ **kw: Any,
+ ) -> None:
if (
checkfirst
or (
) and not self._check_for_name_in_memos(checkfirst, kw):
self.create(bind=bind, checkfirst=checkfirst)
- def _on_table_drop(self, target, bind, checkfirst=False, **kw):
+ def _on_table_drop(
+ self,
+ target: Any,
+ bind: _CreateDropBind,
+ checkfirst: bool = False,
+ **kw: Any,
+ ) -> None:
if (
not self.metadata
and not kw.get("_is_metadata_operation", False)
):
self.drop(bind=bind, checkfirst=checkfirst)
- def _on_metadata_create(self, target, bind, checkfirst=False, **kw):
+ def _on_metadata_create(
+ self,
+ target: Any,
+ bind: _CreateDropBind,
+ checkfirst: bool = False,
+ **kw: Any,
+ ) -> None:
if not self._check_for_name_in_memos(checkfirst, kw):
self.create(bind=bind, checkfirst=checkfirst)
- def _on_metadata_drop(self, target, bind, checkfirst=False, **kw):
+ def _on_metadata_drop(
+ self,
+ target: Any,
+ bind: _CreateDropBind,
+ checkfirst: bool = False,
+ **kw: Any,
+ ) -> None:
if not self._check_for_name_in_memos(checkfirst, kw):
self.drop(bind=bind, checkfirst=checkfirst)
return cls(**kw)
- def create(self, bind=None, checkfirst=True):
+ def create(self, bind: _CreateDropBind, checkfirst: bool = True) -> None:
"""Emit ``CREATE TYPE`` for this
:class:`_postgresql.ENUM`.
super().create(bind, checkfirst=checkfirst)
- def drop(self, bind=None, checkfirst=True):
+ def drop(self, bind: _CreateDropBind, checkfirst: bool = True) -> None:
"""Emit ``DROP TYPE`` for this
:class:`_postgresql.ENUM`.
super().drop(bind, checkfirst=checkfirst)
- def get_dbapi_type(self, dbapi):
+ def get_dbapi_type(self, dbapi: ModuleType) -> None:
"""dont return dbapi.STRING for ENUM in PostgreSQL, since that's
a different type"""
from ..sql._typing import _InfoType
from ..sql.compiler import Compiled
from ..sql.ddl import ExecutableDDLElement
- from ..sql.ddl import SchemaDropper
- from ..sql.ddl import SchemaGenerator
+ from ..sql.ddl import InvokeDDLBase
from ..sql.functions import FunctionElement
from ..sql.schema import DefaultGenerator
from ..sql.schema import HasSchemaAttr
- from ..sql.schema import SchemaItem
+ from ..sql.schema import SchemaVisitable
from ..sql.selectable import TypedReturnsRows
def _run_ddl_visitor(
self,
- visitorcallable: Type[Union[SchemaGenerator, SchemaDropper]],
- element: SchemaItem,
+ visitorcallable: Type[InvokeDDLBase],
+ element: SchemaVisitable,
**kwargs: Any,
) -> None:
"""run a DDL visitor.
options given to the visitor so that "checkfirst" is skipped.
"""
- visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
+ visitorcallable(
+ dialect=self.dialect, connection=self, **kwargs
+ ).traverse_single(element)
class ExceptionContextImpl(ExceptionContext):
def _run_ddl_visitor(
self,
- visitorcallable: Type[Union[SchemaGenerator, SchemaDropper]],
- element: SchemaItem,
+ visitorcallable: Type[InvokeDDLBase],
+ element: SchemaVisitable,
**kwargs: Any,
) -> None:
with self.begin() as conn:
from .interfaces import Dialect
from .url import URL
from ..sql.base import Executable
- from ..sql.ddl import SchemaDropper
- from ..sql.ddl import SchemaGenerator
+ from ..sql.ddl import InvokeDDLBase
from ..sql.schema import HasSchemaAttr
- from ..sql.schema import SchemaItem
+ from ..sql.visitors import Visitable
class MockConnection:
def _run_ddl_visitor(
self,
- visitorcallable: Type[Union[SchemaGenerator, SchemaDropper]],
- element: SchemaItem,
+ visitorcallable: Type[InvokeDDLBase],
+ element: Visitable,
**kwargs: Any,
) -> None:
kwargs["checkfirst"] = False
- visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
+ visitorcallable(
+ dialect=self.dialect, connection=self, **kwargs
+ ).traverse_single(element)
def execute(
self,
from .sql.schema import PrimaryKeyConstraint as PrimaryKeyConstraint
from .sql.schema import SchemaConst as SchemaConst
from .sql.schema import SchemaItem as SchemaItem
+from .sql.schema import SchemaVisitable as SchemaVisitable
from .sql.schema import Sequence as Sequence
from .sql.schema import Table as Table
from .sql.schema import UniqueConstraint as UniqueConstraint
from .sqltypes import TableValueType
from .sqltypes import TupleType
from .type_api import TypeEngine
+ from ..engine import Connection
from ..engine import Dialect
+ from ..engine import Engine
+ from ..engine.mock import MockConnection
from ..util.typing import TypeGuard
_T = TypeVar("_T", bound=Any)
_AutoIncrementType = Union[bool, Literal["auto", "ignore_fk"]]
+_CreateDropBind = Union["Engine", "Connection", "MockConnection"]
+
if TYPE_CHECKING:
def is_sql_compiler(c: Compiled) -> TypeGuard[SQLCompiler]: ...
self.dispatch.after_parent_attach(self, parent)
+class SchemaVisitable(SchemaEventTarget, visitors.Visitable):
+ """Base class for elements that are targets of a :class:`.SchemaVisitor`.
+
+ .. versionadded:: 2.0.41
+
+ """
+
+
class SchemaVisitor(ClauseVisitor):
- """Define the visiting for ``SchemaItem`` objects."""
+ """Define the visiting for ``SchemaItem`` and more
+ generally ``SchemaVisitable`` objects.
+
+ """
__traverse_options__ = {"schema_visitor": True}
class InvokeDDLBase(SchemaVisitor):
- def __init__(self, connection):
+ def __init__(self, connection, **kw):
self.connection = connection
+ assert not kw, f"Unexpected keywords: {kw.keys()}"
@contextlib.contextmanager
def with_ddl_events(self, target, **kw):
from .base import DialectKWArgs
from .base import Executable
from .base import SchemaEventTarget as SchemaEventTarget
+from .base import SchemaVisitable as SchemaVisitable
from .coercions import _document_text_coercion
from .elements import ClauseElement
from .elements import ColumnClause
if typing.TYPE_CHECKING:
from ._typing import _AutoIncrementType
+ from ._typing import _CreateDropBind
from ._typing import _DDLColumnArgument
from ._typing import _DDLColumnReferenceArgument
from ._typing import _InfoType
from ..engine.interfaces import _CoreMultiExecuteParams
from ..engine.interfaces import CoreExecuteOptionsParameter
from ..engine.interfaces import ExecutionContext
- from ..engine.mock import MockConnection
from ..engine.reflection import _ReflectionInfo
from ..sql.selectable import FromClause
_TAB = TypeVar("_TAB", bound="Table")
-_CreateDropBind = Union["Engine", "Connection", "MockConnection"]
-
_ConstraintNameArgument = Optional[Union[str, _NoneName]]
_ServerDefaultArgument = Union[
@inspection._self_inspects
-class SchemaItem(SchemaEventTarget, visitors.Visitable):
+class SchemaItem(SchemaVisitable):
"""Base class for items that define a database schema."""
__visit_name__ = "schema_item"
if TYPE_CHECKING:
from ._typing import _ColumnExpressionArgument
+ from ._typing import _CreateDropBind
from ._typing import _TypeEngineArgument
from .elements import ColumnElement
from .operators import OperatorType
kw.setdefault("_adapted_from", self)
return super().adapt(cls, **kw)
- def create(self, bind, checkfirst=False):
+ def create(self, bind: _CreateDropBind, checkfirst: bool = False) -> None:
"""Issue CREATE DDL for this type, if applicable."""
t = self.dialect_impl(bind.dialect)
if isinstance(t, SchemaType) and t.__class__ is not self.__class__:
t.create(bind, checkfirst=checkfirst)
- def drop(self, bind, checkfirst=False):
+ def drop(self, bind: _CreateDropBind, checkfirst: bool = False) -> None:
"""Issue DROP DDL for this type, if applicable."""
t = self.dialect_impl(bind.dialect)
if isinstance(t, SchemaType) and t.__class__ is not self.__class__:
t.drop(bind, checkfirst=checkfirst)
- def _on_table_create(self, target, bind, **kw):
+ def _on_table_create(
+ self, target: Any, bind: _CreateDropBind, **kw: Any
+ ) -> None:
if not self._is_impl_for_variant(bind.dialect, kw):
return
if isinstance(t, SchemaType) and t.__class__ is not self.__class__:
t._on_table_create(target, bind, **kw)
- def _on_table_drop(self, target, bind, **kw):
+ def _on_table_drop(
+ self, target: Any, bind: _CreateDropBind, **kw: Any
+ ) -> None:
if not self._is_impl_for_variant(bind.dialect, kw):
return
if isinstance(t, SchemaType) and t.__class__ is not self.__class__:
t._on_table_drop(target, bind, **kw)
- def _on_metadata_create(self, target, bind, **kw):
+ def _on_metadata_create(
+ self, target: Any, bind: _CreateDropBind, **kw: Any
+ ) -> None:
if not self._is_impl_for_variant(bind.dialect, kw):
return
if isinstance(t, SchemaType) and t.__class__ is not self.__class__:
t._on_metadata_create(target, bind, **kw)
- def _on_metadata_drop(self, target, bind, **kw):
+ def _on_metadata_drop(
+ self, target: Any, bind: _CreateDropBind, **kw: Any
+ ) -> None:
if not self._is_impl_for_variant(bind.dialect, kw):
return
if isinstance(t, SchemaType) and t.__class__ is not self.__class__:
t._on_metadata_drop(target, bind, **kw)
- def _is_impl_for_variant(self, dialect, kw):
+ def _is_impl_for_variant(
+ self, dialect: Dialect, kw: Dict[str, Any]
+ ) -> Optional[bool]:
variant_mapping = kw.pop("variant_mapping", None)
if not variant_mapping:
# since PostgreSQL is the only DB that has ARRAY this can only
# be integration tested by PG-specific tests
- def _we_are_the_impl(typ):
+ def _we_are_the_impl(typ: SchemaType) -> bool:
return (
typ is self
or isinstance(typ, ARRAY)
return True
elif dialect.name not in variant_mapping:
return _we_are_the_impl(variant_mapping["_default"])
+ else:
+ return None
_EnumTupleArg = Union[Sequence[enum.Enum], Sequence[str]]
"schema",
"metadata",
"name",
+ "dispatch",
):
continue
# assert each value was copied, or that
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
