from __future__ import annotations
import os
-from typing import Callable
from typing import List
from typing import Optional
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from alembic.config import Config
from alembic.script.base import Script
+ from .runtime.environment import ProcessRevisionDirectiveFn
def list_templates(config):
version_path: Optional[str] = None,
rev_id: Optional[str] = None,
depends_on: Optional[str] = None,
- process_revision_directives: Callable = None,
+ process_revision_directives: Optional[ProcessRevisionDirectiveFn] = None,
) -> Union[Optional["Script"], List[Optional["Script"]]]:
"""Create a new revision file.
def merge(
config: "Config",
revisions: str,
- message: str = None,
- branch_label: str = None,
- rev_id: str = None,
+ message: Optional[str] = None,
+ branch_label: Optional[str] = None,
+ rev_id: Optional[str] = None,
) -> Optional["Script"]:
"""Merge two revisions together. Creates a new migration file.
stdout: TextIO = sys.stdout,
cmd_opts: Optional[Namespace] = None,
config_args: util.immutabledict = util.immutabledict(),
- attributes: dict = None,
+ attributes: Optional[dict] = None,
) -> None:
"""Construct a new :class:`.Config`"""
self.config_file_name = file_
from sqlalchemy.sql.schema import MetaData
from .config import Config
+ from .operations import MigrateOperation
from .runtime.migration import _ProxyTransaction
from .runtime.migration import MigrationContext
from .script import ScriptDirectory
-
### end imports ###
-def begin_transaction() -> Union["_ProxyTransaction", ContextManager]:
+def begin_transaction() -> Union[_ProxyTransaction, ContextManager]:
"""Return a context manager that will
enclose an operation within a "transaction",
as defined by the environment's offline
connection: Optional[Connection] = None,
url: Optional[str] = None,
dialect_name: Optional[str] = None,
- dialect_opts: Optional[dict] = None,
+ dialect_opts: Optional[Dict[str, Any]] = None,
transactional_ddl: Optional[bool] = None,
transaction_per_migration: bool = False,
output_buffer: Optional[TextIO] = None,
starting_rev: Optional[str] = None,
tag: Optional[str] = None,
- template_args: Optional[dict] = None,
+ template_args: Optional[Dict[str, Any]] = None,
render_as_batch: bool = False,
target_metadata: Optional[MetaData] = None,
- include_name: Optional[Callable] = None,
- include_object: Optional[Callable] = None,
+ include_name: Optional[Callable[..., bool]] = None,
+ include_object: Optional[Callable[..., bool]] = None,
include_schemas: bool = False,
- process_revision_directives: Optional[Callable] = None,
+ process_revision_directives: Optional[
+ Callable[
+ [MigrationContext, Tuple[str, str], List[MigrateOperation]], None
+ ]
+ ] = None,
compare_type: bool = False,
compare_server_default: bool = False,
- render_item: Optional[Callable] = None,
+ render_item: Optional[Callable[..., bool]] = None,
literal_binds: bool = False,
upgrade_token: str = "upgrades",
downgrade_token: str = "downgrades",
alembic_module_prefix: str = "op.",
sqlalchemy_module_prefix: str = "sa.",
user_module_prefix: Optional[str] = None,
- on_version_apply: Optional[Callable] = None,
+ on_version_apply: Optional[Callable[..., None]] = None,
**kw: Any,
) -> None:
"""Configure a :class:`.MigrationContext` within this
from .operations.ops import BatchOperations
from .operations.ops import MigrateOperation
+ from .runtime.migration import MigrationContext
from .util.sqla_compat import _literal_bindparam
-
### end imports ###
def add_column(
"""
-def get_context():
+def get_context() -> MigrationContext:
"""Return the :class:`.MigrationContext` object that's
currently in use.
"""
-def implementation_for(op_cls: Any) -> Callable:
+def implementation_for(op_cls: Any) -> Callable[..., Any]:
"""Register an implementation for a given :class:`.MigrateOperation`.
This is part of the operation extensibility API.
def inline_literal(
value: Union[str, int], type_: None = None
-) -> "_literal_bindparam":
+) -> _literal_bindparam:
"""Produce an 'inline literal' expression, suitable for
using in an INSERT, UPDATE, or DELETE statement.
def register_operation(
name: str, sourcename: Optional[str] = None
-) -> Callable:
+) -> Callable[..., Any]:
"""Register a new operation for this class.
This method is normally used to add new operations
from ..util.compat import formatannotation_fwdref
from ..util.compat import inspect_formatargspec
from ..util.compat import inspect_getfullargspec
+from ..util.sqla_compat import _literal_bindparam
NoneType = type(None)
from .ops import MigrateOperation
from ..ddl import DefaultImpl
from ..runtime.migration import MigrationContext
- from ..util.sqla_compat import _literal_bindparam
__all__ = ("Operations", "BatchOperations")
def __init__(
self,
- migration_context: "MigrationContext",
- impl: Optional["BatchOperationsImpl"] = None,
+ migration_context: MigrationContext,
+ impl: Optional[BatchOperationsImpl] = None,
) -> None:
"""Construct a new :class:`.Operations`
@classmethod
def register_operation(
cls, name: str, sourcename: Optional[str] = None
- ) -> Callable:
+ ) -> Callable[..., Any]:
"""Register a new operation for this class.
This method is normally used to add new operations
return register
@classmethod
- def implementation_for(cls, op_cls: Any) -> Callable:
+ def implementation_for(cls, op_cls: Any) -> Callable[..., Any]:
"""Register an implementation for a given :class:`.MigrateOperation`.
This is part of the operation extensibility API.
@classmethod
@contextmanager
def context(
- cls, migration_context: "MigrationContext"
- ) -> Iterator["Operations"]:
+ cls, migration_context: MigrationContext
+ ) -> Iterator[Operations]:
op = Operations(migration_context)
op._install_proxy()
yield op
yield batch_op
impl.flush()
- def get_context(self):
+ def get_context(self) -> MigrationContext:
"""Return the :class:`.MigrationContext` object that's
currently in use.
return self.migration_context
- def invoke(self, operation: "MigrateOperation") -> Any:
+ def invoke(self, operation: MigrateOperation) -> Any:
"""Given a :class:`.MigrateOperation`, invoke it in terms of
this :class:`.Operations` instance.
)
return fn(self, operation)
- def f(self, name: str) -> "conv":
+ def f(self, name: str) -> conv:
"""Indicate a string name that has already had a naming convention
applied to it.
def inline_literal(
self, value: Union[str, int], type_: None = None
- ) -> "_literal_bindparam":
+ ) -> _literal_bindparam:
r"""Produce an 'inline literal' expression, suitable for
using in an INSERT, UPDATE, or DELETE statement.
"""
return sqla_compat._literal_bindparam(None, value, type_=type_)
- def get_bind(self) -> "Connection":
+ def get_bind(self) -> Connection:
"""Return the current 'bind'.
Under normal circumstances, this is the
from typing import TYPE_CHECKING
from typing import Union
+from .migration import _ProxyTransaction
from .migration import MigrationContext
from .. import util
from ..operations import Operations
from sqlalchemy.sql.elements import ClauseElement
from sqlalchemy.sql.schema import MetaData
- from .migration import _ProxyTransaction
from ..config import Config
from ..ddl import DefaultImpl
+ from ..operations.ops import MigrateOperation
from ..script.base import ScriptDirectory
_RevNumber = Optional[Union[str, Tuple[str, ...]]]
+ProcessRevisionDirectiveFn = Callable[
+ [MigrationContext, Tuple[str, str], List["MigrateOperation"]], None
+]
+
class EnvironmentContext(util.ModuleClsProxy):
"""
def __init__(
- self, config: "Config", script: "ScriptDirectory", **kw: Any
+ self, config: Config, script: ScriptDirectory, **kw: Any
) -> None:
r"""Construct a new :class:`.EnvironmentContext`.
self.script = script
self.context_opts = kw
- def __enter__(self) -> "EnvironmentContext":
+ def __enter__(self) -> EnvironmentContext:
"""Establish a context which provides a
:class:`.EnvironmentContext` object to
env.py scripts.
@overload
def get_x_argument( # type:ignore[misc]
- self, as_dictionary: "Literal[False]" = ...
+ self, as_dictionary: Literal[False] = ...
) -> List[str]:
...
@overload
def get_x_argument( # type:ignore[misc]
- self, as_dictionary: "Literal[True]" = ...
+ self, as_dictionary: Literal[True] = ...
) -> Dict[str, str]:
...
def configure(
self,
- connection: Optional["Connection"] = None,
+ connection: Optional[Connection] = None,
url: Optional[str] = None,
dialect_name: Optional[str] = None,
- dialect_opts: Optional[dict] = None,
+ dialect_opts: Optional[Dict[str, Any]] = None,
transactional_ddl: Optional[bool] = None,
transaction_per_migration: bool = False,
output_buffer: Optional[TextIO] = None,
starting_rev: Optional[str] = None,
tag: Optional[str] = None,
- template_args: Optional[dict] = None,
+ template_args: Optional[Dict[str, Any]] = None,
render_as_batch: bool = False,
- target_metadata: Optional["MetaData"] = None,
- include_name: Optional[Callable] = None,
- include_object: Optional[Callable] = None,
+ target_metadata: Optional[MetaData] = None,
+ include_name: Optional[Callable[..., bool]] = None,
+ include_object: Optional[Callable[..., bool]] = None,
include_schemas: bool = False,
- process_revision_directives: Optional[Callable] = None,
+ process_revision_directives: Optional[
+ ProcessRevisionDirectiveFn
+ ] = None,
compare_type: bool = False,
compare_server_default: bool = False,
- render_item: Optional[Callable] = None,
+ render_item: Optional[Callable[..., bool]] = None,
literal_binds: bool = False,
upgrade_token: str = "upgrades",
downgrade_token: str = "downgrades",
alembic_module_prefix: str = "op.",
sqlalchemy_module_prefix: str = "sa.",
user_module_prefix: Optional[str] = None,
- on_version_apply: Optional[Callable] = None,
+ on_version_apply: Optional[Callable[..., None]] = None,
**kw: Any,
) -> None:
"""Configure a :class:`.MigrationContext` within this
def execute(
self,
- sql: Union["ClauseElement", str],
+ sql: Union[ClauseElement, str],
execution_options: Optional[dict] = None,
) -> None:
"""Execute the given SQL using the current change context.
def begin_transaction(
self,
- ) -> Union["_ProxyTransaction", ContextManager]:
+ ) -> Union[_ProxyTransaction, ContextManager]:
"""Return a context manager that will
enclose an operation within a "transaction",
as defined by the environment's offline
return self.get_context().begin_transaction()
- def get_context(self) -> "MigrationContext":
+ def get_context(self) -> MigrationContext:
"""Return the current :class:`.MigrationContext` object.
If :meth:`.EnvironmentContext.configure` has not been
raise Exception("No context has been configured yet.")
return self._migration_context
- def get_bind(self) -> "Connection":
+ def get_bind(self) -> Connection:
"""Return the current 'bind'.
In "online" mode, this is the
"""
return self.get_context().bind # type: ignore[return-value]
- def get_impl(self) -> "DefaultImpl":
+ def get_impl(self) -> DefaultImpl:
return self.get_context().impl
@overload
-def to_tuple(x: None, default: _T = None) -> _T:
+def to_tuple(x: None, default: Optional[_T] = None) -> _T:
...
--- /dev/null
+.. change::
+ :tags: bug, typing
+ :tickets: 1110
+
+ Fixed typing issue where :paramref:`.revision.process_revision_directives`
+ was not fully typed; additionally ensured all ``Callable`` and ``Dict``
+ arguments to :meth:`.EnvironmentContext.configure` include parameters in
+ the typing declaration.
+
+ Additionally updated the codebase for Mypy 0.990 compliance.
\ No newline at end of file
def _generate_stub_for_attr(cls, name, printer, env):
try:
annotations = typing.get_type_hints(cls, env)
- except NameError as e:
+ except NameError:
annotations = cls.__annotations__
type_ = annotations.get(name, "Any")
if isinstance(type_, str) and type_[0] in "'\"":
if getattr(annotation, "__module__", None) == "typing":
retval = repr(annotation).replace("typing.", "")
elif isinstance(annotation, type):
- if annotation.__module__ in ("builtins", base_module):
- retval = annotation.__qualname__
- else:
- retval = annotation.__module__ + "." + annotation.__qualname__
+ retval = annotation.__qualname__
else:
retval = annotation
'''{fn.__doc__}'''
"""
)
+
printer.write_indented_block(func_text)
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
