recursive-include doc *.html *.css *.txt *.js *.png *.py Makefile *.rst *.sty
recursive-include examples *.py *.xml
-recursive-include test *.py *.dat
+recursive-include test *.py *.dat *.testpatch
# include the c extensions, which otherwise
# don't come in if --with-cextensions isn't specified.
--- /dev/null
+.. change::
+ :tags: bug, mypy
+ :tickets: sqlalchemy/sqlalchemy2-stubs/#14
+
+ Fixed issue in mypy plugin where newly added support for
+ :func:`_orm.as_declarative` needed to more fully add the
+ ``DeclarativeMeta`` class to the mypy interpreter's state so that it does
+ not result in a name not found error; additionally improves how global
+ names are setup for the plugin including the ``Mapped`` name.
+
left_hand_explicit_type = AnyType(TypeOfAny.special_form)
- descriptor = api.modules["sqlalchemy.orm.attributes"].names["Mapped"]
+ descriptor = api.lookup("__sa_Mapped", cls)
left_node = NameExpr(stmt.var.name)
left_node.node = stmt.var
# <attr> : Mapped[<typ>] =
# _sa_Mapped._empty_constructor(lambda: <function body>)
# the function body is maintained so it gets type checked internally
- api.add_symbol_table_node("_sa_Mapped", descriptor)
- column_descriptor = nodes.NameExpr("_sa_Mapped")
+ column_descriptor = nodes.NameExpr("__sa_Mapped")
column_descriptor.fullname = "sqlalchemy.orm.Mapped"
mm = nodes.MemberExpr(column_descriptor, "_empty_constructor")
"""
if not is_subtype(left_hand_explicit_type, python_type_for_type):
- descriptor = api.modules["sqlalchemy.orm.attributes"].names["Mapped"]
+ descriptor = api.lookup("__sa_Mapped", node)
effective_type = Instance(descriptor.node, [python_type_for_type])
)
util.fail(api, msg.format(node.name), node)
- descriptor = api.modules["sqlalchemy.orm.attributes"].names["Mapped"]
-
+ descriptor = api.lookup("__sa_Mapped", node)
return Instance(descriptor.node, [AnyType(TypeOfAny.special_form)])
else:
name: typ for name, typ in cls_metadata.mapped_attr_names
}
- descriptor = api.modules["sqlalchemy.orm.attributes"].names["Mapped"]
-
+ descriptor = api.lookup("__sa_Mapped", cls)
for stmt in cls.defs.body:
# for a re-apply, all of our statements are AssignmentStmt;
# @declared_attr calls will have been converted and this
attrname : Mapped[Optional[int]] = <meaningless temp node>
"""
- descriptor = api.modules["sqlalchemy.orm.attributes"].names["Mapped"]
-
+ descriptor = api.lookup("__sa_Mapped", stmt)
left_node = lvalue.node
inst = Instance(descriptor.node, [python_type_for_type])
# and "registry.as_declarative_base()" methods.
# this seems like a bug in mypy that these decorators are otherwise
# skipped.
+
if (
isinstance(decorator, nodes.CallExpr)
and isinstance(decorator.callee, nodes.MemberExpr)
)
+def _add_globals(ctx: ClassDefContext):
+ """Add __sa_DeclarativeMeta and __sa_Mapped symbol to the global space
+ for all class defs
+
+ """
+
+ util.add_global(
+ ctx,
+ "sqlalchemy.orm.decl_api",
+ "DeclarativeMeta",
+ "__sa_DeclarativeMeta",
+ )
+
+ util.add_global(ctx, "sqlalchemy.orm.attributes", "Mapped", "__sa_Mapped")
+
+
def _cls_metadata_hook(ctx: ClassDefContext) -> None:
+ _add_globals(ctx)
decl_class._scan_declarative_assignments_and_apply_types(ctx.cls, ctx.api)
def _base_cls_hook(ctx: ClassDefContext) -> None:
+ _add_globals(ctx)
decl_class._scan_declarative_assignments_and_apply_types(ctx.cls, ctx.api)
def _cls_decorator_hook(ctx: ClassDefContext) -> None:
+ _add_globals(ctx)
assert isinstance(ctx.reason, nodes.MemberExpr)
expr = ctx.reason.expr
assert names._type_id_for_named_node(expr.node.type.type) is names.REGISTRY
decl_class._scan_declarative_assignments_and_apply_types(ctx.cls, ctx.api)
-def _make_declarative_meta(
- api: SemanticAnalyzerPluginInterface, target_cls: ClassDef
-):
- declarative_meta_sym: SymbolTableNode = api.modules[
- "sqlalchemy.orm.decl_api"
- ].names["DeclarativeMeta"]
- declarative_meta_typeinfo: TypeInfo = declarative_meta_sym.node
-
- declarative_meta_name: NameExpr = NameExpr("DeclarativeMeta")
- declarative_meta_name.kind = GDEF
- declarative_meta_name.fullname = "sqlalchemy.orm.decl_api.DeclarativeMeta"
- declarative_meta_name.node = declarative_meta_typeinfo
-
- target_cls.metaclass = declarative_meta_name
-
- declarative_meta_instance = Instance(declarative_meta_typeinfo, [])
-
- info = target_cls.info
- info.declared_metaclass = info.metaclass_type = declarative_meta_instance
-
-
def _base_cls_decorator_hook(ctx: ClassDefContext) -> None:
+ _add_globals(ctx)
cls = ctx.cls
"""Generate a declarative Base class when the declarative_base() function
is encountered."""
+ _add_globals(ctx)
+
cls = ClassDef(ctx.name, Block([]))
cls.fullname = ctx.api.qualified_name(ctx.name)
info.fallback_to_any = True
ctx.api.add_symbol_table_node(ctx.name, SymbolTableNode(GDEF, info))
+
+
+def _make_declarative_meta(
+ api: SemanticAnalyzerPluginInterface, target_cls: ClassDef
+):
+
+ declarative_meta_name: NameExpr = NameExpr("__sa_DeclarativeMeta")
+ declarative_meta_name.kind = GDEF
+ declarative_meta_name.fullname = "sqlalchemy.orm.decl_api.DeclarativeMeta"
+
+ # installed by _add_globals
+ sym = api.lookup("__sa_DeclarativeMeta", target_cls)
+
+ declarative_meta_typeinfo = sym.node
+ declarative_meta_name.node = declarative_meta_typeinfo
+
+ target_cls.metaclass = declarative_meta_name
+
+ declarative_meta_instance = Instance(declarative_meta_typeinfo, [])
+
+ info = target_cls.info
+ info.declared_metaclass = info.metaclass_type = declarative_meta_instance
return api.fail(msg, ctx)
+def add_global(
+ ctx: SemanticAnalyzerPluginInterface,
+ module: str,
+ symbol_name: str,
+ asname: str,
+):
+ module_globals = ctx.api.modules[ctx.api.cur_mod_id].names
+
+ if asname not in module_globals:
+ lookup_sym: SymbolTableNode = ctx.api.modules[module].names[
+ symbol_name
+ ]
+
+ module_globals[asname] = lookup_sym
+
+
def _get_callexpr_kwarg(callexpr: CallExpr, name: str) -> Optional[NameExpr]:
try:
arg_idx = callexpr.arg_names.index(name)
lambda: util.cpython, "cPython interpreter needed"
)
+ @property
+ def patch_library(self):
+ def check_lib():
+ try:
+ __import__("patch")
+ except ImportError:
+ return False
+ else:
+ return True
+
+ return exclusions.only_if(check_lib, "patch library needed")
+
@property
def non_broken_pickle(self):
from sqlalchemy.util import pickle
+from typing import List
+from typing import Optional
+
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.ext.declarative import as_declarative
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import relationship
+from sqlalchemy.sql.schema import ForeignKey
@as_declarative
class Foo(Base):
__tablename__ = "foo"
id: int = Column(Integer(), primary_key=True)
- name: str = Column(String)
+ name: Mapped[str] = Column(String)
+
+ bar: List["Bar"] = relationship("Bar")
+
+
+class Bar(Base):
+ __tablename__ = "bar"
+ id: int = Column(Integer(), primary_key=True)
+ foo_id: int = Column(ForeignKey("foo.id"))
+
+ foo: Optional[Foo] = relationship(Foo)
f1 = Foo()
--- /dev/null
+from typing import TYPE_CHECKING
+
+from sqlalchemy import Column
+from sqlalchemy import Integer
+from sqlalchemy.orm import as_declarative
+from sqlalchemy.orm import declared_attr
+from sqlalchemy.orm import Mapped
+from .address import Address
+from .user import User
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm.decl_api import DeclarativeMeta
+
+
+@as_declarative()
+class Base(object):
+ @declared_attr
+ def __tablename__(self) -> Mapped[str]:
+ return self.__name__.lower()
+
+ id = Column(Integer, primary_key=True)
+
+
+__all__ = ["User", "Address"]
--- /dev/null
+from typing import TYPE_CHECKING
+
+from . import Base
+from .user import HasUser
+
+if TYPE_CHECKING:
+ from .user import User # noqa
+ from sqlalchemy import Integer, Column # noqa
+ from sqlalchemy.orm import RelationshipProperty # noqa
+
+
+class Address(Base, HasUser):
+ pass
--- /dev/null
+diff --git a/test/ext/mypy/incremental/stubs_14/user.py b/test/ext/mypy/incremental/stubs_14/user.py
+index 2c60403e4..c7e8f8874 100644
+--- a/user.py
++++ b/user.py
+@@ -18,6 +18,8 @@ if TYPE_CHECKING:
+ class User(Base):
+ name = Column(String)
+
++ othername = Column(String)
++
+ addresses: Mapped[List["Address"]] = relationship(
+ "Address", back_populates="user"
+ )
--- /dev/null
+from typing import List
+from typing import TYPE_CHECKING
+
+from sqlalchemy import Column
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
+from sqlalchemy import String
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import relationship
+from sqlalchemy.orm.decl_api import declared_attr
+from sqlalchemy.orm.relationships import RelationshipProperty
+from . import Base
+
+if TYPE_CHECKING:
+ from .address import Address
+
+
+class User(Base):
+ name = Column(String)
+
+ othername = Column(String)
+
+ addresses: Mapped[List["Address"]] = relationship(
+ "Address", back_populates="user"
+ )
+
+
+class HasUser:
+ @declared_attr
+ def user_id(self) -> "Column[Integer]":
+ return Column(
+ Integer,
+ ForeignKey(User.id, ondelete="CASCADE", onupdate="CASCADE"),
+ nullable=False,
+ )
+
+ @declared_attr
+ def user(self) -> RelationshipProperty[User]:
+ return relationship(User)
import os
import re
+import shutil
import tempfile
from sqlalchemy import testing
def mypy_runner(self, cachedir):
from mypy import api
- def run(filename, use_plugin=True):
- path = os.path.join(os.path.dirname(__file__), "files", filename)
+ def run(
+ filename, use_plugin=True, incremental=False, working_dir=None
+ ):
+ if working_dir:
+ path = os.path.join(working_dir, filename)
+ else:
+ path = os.path.join(
+ os.path.dirname(__file__), "files", filename
+ )
args = [
"--strict",
return run
+ def _incremental_dirs():
+ path = os.path.join(os.path.dirname(__file__), "incremental")
+ return [
+ d for d in os.listdir(path) if os.path.isdir(os.path.join(path, d))
+ ]
+
+ @testing.combinations(
+ *[(dirname,) for dirname in _incremental_dirs()], argnames="dirname"
+ )
+ @testing.requires.patch_library
+ def test_incremental(self, mypy_runner, cachedir, dirname):
+ import patch
+
+ path = os.path.join(os.path.dirname(__file__), "incremental", dirname)
+ dest = os.path.join(cachedir, "mymodel")
+ os.mkdir(dest)
+
+ patches = set()
+
+ print("incremental test: %s" % dirname)
+
+ for fname in os.listdir(path):
+ if fname.endswith(".py"):
+ shutil.copy(
+ os.path.join(path, fname), os.path.join(dest, fname)
+ )
+ print("copying to: %s" % os.path.join(dest, fname))
+ elif fname.endswith(".testpatch"):
+ patches.add(fname)
+
+ for patchfile in [None] + sorted(patches):
+ if patchfile is not None:
+ print("Applying patchfile %s" % patchfile)
+ patch_obj = patch.fromfile(os.path.join(path, patchfile))
+ patch_obj.apply(1, dest)
+ print("running mypy against %s/mymodel" % cachedir)
+ result = mypy_runner(
+ "mymodel",
+ use_plugin=True,
+ incremental=True,
+ working_dir=cachedir,
+ )
+ eq_(
+ result[2],
+ 0,
+ msg="Failure after applying patch %s: %s"
+ % (patchfile, result[0]),
+ )
+
def _file_combinations():
path = os.path.join(os.path.dirname(__file__), "files")
return [f for f in os.listdir(path) if f.endswith(".py")]
mock; python_version < '3.3'
importlib_metadata; python_version < '3.8'
mypy
+ patch==1.*
git+https://github.com/sqlalchemy/sqlalchemy2-stubs
commands =
pytest test/ext/mypy/test_mypy_plugin_py3k.py {posargs}
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
